GH-33823: [C++][IPC] Improve error messages when opening files that are the wrong format

cpp/src/arrow/ipc/message.cc

@@ -1051,6 +1051,19 @@ class InputStreamMessageReader : public MessageReader, public MessageDecoderList
 
   Status OnMessageDecoded(std::unique_ptr<Message> 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<io::InputStream> owned_stream_;
   std::unique_ptr<Message> message_;
+  bool first_message_ = true;
   MessageDecoder decoder_;
 };
 

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<RecordBatch> 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<Buffer>(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<RecordBatch> 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<RecordBatch> 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<io::BufferReader>(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) {}

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<const char*>(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(