diff --git a/cpp/src/arrow/ipc/message.cc b/cpp/src/arrow/ipc/message.cc index 4123a6b1735..38780d7572c 100644 --- a/cpp/src/arrow/ipc/message.cc +++ b/cpp/src/arrow/ipc/message.cc @@ -87,6 +87,7 @@ class Message::MessageImpl { : metadata_(std::move(metadata)), message_(nullptr), body_(std::move(body)) {} Status Open() { + RETURN_NOT_OK(MaybeAlignMetadata(&metadata_)); RETURN_NOT_OK( internal::VerifyMessage(metadata_->data(), metadata_->size(), &message_)); diff --git a/cpp/src/arrow/ipc/read_write_test.cc b/cpp/src/arrow/ipc/read_write_test.cc index cfa6a541d5c..51530c2bc8e 100644 --- a/cpp/src/arrow/ipc/read_write_test.cc +++ b/cpp/src/arrow/ipc/read_write_test.cc @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -224,6 +225,31 @@ TEST(TestMessage, Verify) { ASSERT_FALSE(message.Verify()); } +TEST(TestMessage, OpenUnalignedMetadata) { + auto custom_metadata = key_value_metadata({"foo"}, {"bar"}); + ASSERT_OK_AND_ASSIGN( + auto serialized, + internal::WriteRecordBatchMessage( + /*length=*/0, /*body_length=*/0, custom_metadata, + /*nodes=*/{}, + /*buffers=*/{}, /*variadic_counts=*/{}, IpcWriteOptions::Defaults())); + + // Place the metadata at an odd address, as when it is sliced out of a larger + // buffer (e.g. a Flight data_header) + ASSERT_OK_AND_ASSIGN(std::shared_ptr storage, + AllocateBuffer(serialized->size() + 1)); + std::memcpy(storage->mutable_data() + 1, serialized->data(), serialized->size()); + auto unaligned = SliceBuffer(storage, 1, serialized->size()); + ASSERT_NE(0, reinterpret_cast(unaligned->data()) % 8); + + ASSERT_OK_AND_ASSIGN(auto message, Message::Open(unaligned, /*body=*/nullptr)); + ASSERT_EQ(0, reinterpret_cast(message->metadata()->data()) % 8); + ASSERT_TRUE(message->metadata()->Equals(*serialized)); + ASSERT_EQ(MessageType::RECORD_BATCH, message->type()); + ASSERT_EQ(0, message->body_length()); + ASSERT_TRUE(message->custom_metadata()->Equals(*custom_metadata)); +} + INSTANTIATE_TEST_SUITE_P(TestMessage, TestMessage, ::testing::ValuesIn(kMetadataVersions));