diff --git a/cpp/src/arrow/ipc/message.cc b/cpp/src/arrow/ipc/message.cc index 84ee62fe9e8d..b797cb6486cb 100644 --- a/cpp/src/arrow/ipc/message.cc +++ b/cpp/src/arrow/ipc/message.cc @@ -1051,6 +1051,19 @@ class InputStreamMessageReader : public MessageReader, public MessageDecoderList Status OnMessageDecoded(std::unique_ptr message) override { message_ = std::move(message); + first_message_ = false; + return Status::OK(); + } + + Status OnMetadata() override { + // An IPC file starts with "ARROW1". Read as a stream, its first four bytes + // ("ARRO") decode as a metadata length of 0x4F525241. + constexpr int64_t kArrowMagicAsMetadataLength = 0x4F525241; + if (first_message_ && decoder_.next_required_size() == kArrowMagicAsMetadataLength) { + return Status::Invalid( + "This appears to be an Arrow IPC file. " + "Try the IPC file reader instead of the IPC stream reader."); + } return Status::OK(); } @@ -1063,6 +1076,7 @@ class InputStreamMessageReader : public MessageReader, public MessageDecoderList io::InputStream* stream_; std::shared_ptr owned_stream_; std::unique_ptr message_; + bool first_message_ = true; MessageDecoder decoder_; }; diff --git a/cpp/src/arrow/ipc/read_write_test.cc b/cpp/src/arrow/ipc/read_write_test.cc index 15cf0258b2ee..ba4f8c87588a 100644 --- a/cpp/src/arrow/ipc/read_write_test.cc +++ b/cpp/src/arrow/ipc/read_write_test.cc @@ -2265,6 +2265,75 @@ TEST(TestRecordBatchStreamReader, MalformedInput) { ASSERT_RAISES(Invalid, RecordBatchStreamReader::Open(&garbage_reader)); } +TEST(TestRecordBatchStreamReader, OpenFileFormatSuggestsFileReader) { + std::shared_ptr batch; + ASSERT_OK(MakeIntRecordBatch(&batch)); + + FileWriterHelper helper; + ASSERT_OK(helper.Init(batch->schema(), IpcWriteOptions::Defaults())); + ASSERT_OK(helper.WriteBatch(batch)); + ASSERT_OK(helper.Finish()); + + io::BufferReader reader(helper.buffer_); + // Check we mention using the file_reader when we detect file format + EXPECT_RAISES_WITH_MESSAGE_THAT(Invalid, + ::testing::HasSubstr("Try the IPC file reader"), + RecordBatchStreamReader::Open(&reader)); +} + +TEST(TestRecordBatchStreamReader, CorruptDataDoesNotSuggestFileReader) { + // Continuation marker + metadata_length = 100, then 8 bytes of non-magic data. + const std::string corrupt( + "\xff\xff\xff\xff" + "\x64\x00\x00\x00" + "ABABABAB", + 16); + auto buffer = std::make_shared(corrupt); + io::BufferReader reader(buffer); + // Validate that we don't suggest file reader when file is just corrupt + EXPECT_RAISES_WITH_MESSAGE_THAT( + Invalid, ::testing::Not(::testing::HasSubstr("Try the IPC file reader")), + RecordBatchStreamReader::Open(&reader)); +} + +TEST(TestRecordBatchStreamReader, FileMagicAfterFirstMessageDoesNotSuggestFileReader) { + std::shared_ptr batch; + ASSERT_OK(MakeIntRecordBatch(&batch)); + + StreamWriterHelper helper; + ASSERT_OK(helper.Init(batch->schema(), IpcWriteOptions::Defaults())); + ASSERT_OK(helper.WriteBatch(batch)); + ASSERT_OK(helper.Finish()); + + // Replace the 8-byte marker with the IPC file magic. + std::string data = helper.buffer_->ToString(); + data.resize(data.size() - 8); + data.append("ARROW1\0\0", 8); + io::BufferReader reader(Buffer::FromString(std::move(data))); + + ASSERT_OK_AND_ASSIGN(auto stream_reader, RecordBatchStreamReader::Open(&reader)); + ASSERT_OK(stream_reader->Next()); + EXPECT_RAISES_WITH_MESSAGE_THAT( + Invalid, ::testing::Not(::testing::HasSubstr("Try the IPC file reader")), + stream_reader->Next()); +} + +TEST(TestRecordBatchFileReader, OpenStreamFormatSuggestsStreamReader) { + std::shared_ptr batch; + ASSERT_OK(MakeIntRecordBatch(&batch)); + + StreamWriterHelper helper; + ASSERT_OK(helper.Init(batch->schema(), IpcWriteOptions::Defaults())); + ASSERT_OK(helper.WriteBatch(batch)); + ASSERT_OK(helper.Finish()); + + auto buf_reader = std::make_shared(helper.buffer_); + // Check we mention using the stream_reader when we detect stream format + EXPECT_RAISES_WITH_MESSAGE_THAT( + Invalid, ::testing::HasSubstr("use the IPC stream reader"), + RecordBatchFileReader::Open(buf_reader.get(), helper.buffer_->size())); +} + class EndlessCollectListener : public CollectListener { public: EndlessCollectListener() : CollectListener(), decoder_(nullptr) {} diff --git a/cpp/src/arrow/ipc/reader.cc b/cpp/src/arrow/ipc/reader.cc index 580081384308..68b0d1068c2c 100644 --- a/cpp/src/arrow/ipc/reader.cc +++ b/cpp/src/arrow/ipc/reader.cc @@ -1890,7 +1890,9 @@ class RecordBatchFileReaderImpl : public RecordBatchFileReader { const auto magic_start = buffer->data() + sizeof(int32_t); if (std::string_view(reinterpret_cast(magic_start), kMagicSize) != kArrowMagicBytes) { - return Status::Invalid("Not an Arrow file"); + return Status::Invalid( + "Not an Arrow file. If this is an Arrow IPC stream, use " + "the IPC stream reader instead."); } int32_t footer_length = bit_util::FromLittleEndian(