ARROW-5559: [C++] Add an IpcOptions structure

Switch existing APIs to take IpcOptions rather than individual option arguments.

Closes #5032 from pitrou/ARROW-5559-ipc-options and squashes the following commits:

e39ddc6 <Antoine Pitrou> ARROW-5559:  Add an IpcOptions structure

Authored-by: Antoine Pitrou <antoine@python.org>
Signed-off-by: Antoine Pitrou <antoine@python.org>

Author: pitrou

cpp/src/arrow/CMakeLists.txt

@@ -255,6 +255,7 @@ if(ARROW_IPC)
       ipc/json-simple.cc
       ipc/message.cc
       ipc/metadata-internal.cc
+      ipc/options.cc
       ipc/reader.cc
       ipc/writer.cc)
   set(ARROW_SRCS ${ARROW_SRCS} ${ARROW_IPC_SRCS})

cpp/src/arrow/extension_type-test.cc

@@ -221,7 +221,8 @@ auto RoundtripBatch = [](const std::shared_ptr<RecordBatch>& batch,
                          std::shared_ptr<RecordBatch>* out) {
   std::shared_ptr<io::BufferOutputStream> out_stream;
   ASSERT_OK(io::BufferOutputStream::Create(1024, default_memory_pool(), &out_stream));
-  ASSERT_OK(ipc::WriteRecordBatchStream({batch}, out_stream.get()));
+  ASSERT_OK(ipc::WriteRecordBatchStream({batch}, ipc::IpcOptions::Defaults(),
+                                        out_stream.get()));
 
   std::shared_ptr<Buffer> complete_ipc_stream;
   ASSERT_OK(out_stream->Finish(&complete_ipc_stream));
@@ -273,7 +274,8 @@ TEST_F(TestExtensionType, UnrecognizedExtension) {
   // and ensure that a plain instance of the storage type is created
   std::shared_ptr<io::BufferOutputStream> out_stream;
   ASSERT_OK(io::BufferOutputStream::Create(1024, default_memory_pool(), &out_stream));
-  ASSERT_OK(ipc::WriteRecordBatchStream({batch}, out_stream.get()));
+  ASSERT_OK(ipc::WriteRecordBatchStream({batch}, ipc::IpcOptions::Defaults(),
+                                        out_stream.get()));
 
   std::shared_ptr<Buffer> complete_ipc_stream;
   ASSERT_OK(out_stream->Finish(&complete_ipc_stream));

cpp/src/arrow/flight/client.cc

@@ -251,13 +251,13 @@ class GrpcStreamWriter : public FlightStreamWriter {
       std::shared_ptr<grpc::ClientReaderWriter<pb::FlightData, pb::PutResult>> writer,
       std::unique_ptr<FlightStreamWriter>* out);
 
-  Status WriteRecordBatch(const RecordBatch& batch, bool allow_64bit = false) override {
-    return WriteWithMetadata(batch, nullptr, allow_64bit);
+  Status WriteRecordBatch(const RecordBatch& batch) override {
+    return WriteWithMetadata(batch, nullptr);
   }
-  Status WriteWithMetadata(const RecordBatch& batch, std::shared_ptr<Buffer> app_metadata,
-                           bool allow_64bit = false) override {
+  Status WriteWithMetadata(const RecordBatch& batch,
+                           std::shared_ptr<Buffer> app_metadata) override {
     app_metadata_ = app_metadata;
-    return batch_writer_->WriteRecordBatch(batch, allow_64bit);
+    return batch_writer_->WriteRecordBatch(batch);
   }
   void set_memory_pool(MemoryPool* pool) override {
     batch_writer_->set_memory_pool(pool);

cpp/src/arrow/flight/client.h

@@ -86,8 +86,7 @@ class ARROW_FLIGHT_EXPORT FlightStreamReader : public MetadataRecordBatchReader
 class ARROW_FLIGHT_EXPORT FlightStreamWriter : public ipc::RecordBatchWriter {
  public:
   virtual Status WriteWithMetadata(const RecordBatch& batch,
-                                   std::shared_ptr<Buffer> app_metadata,
-                                   bool allow_64bit = false) = 0;
+                                   std::shared_ptr<Buffer> app_metadata) = 0;
 };
 
 #ifdef _MSC_VER

cpp/src/arrow/flight/perf-server.cc

@@ -74,7 +74,7 @@ class PerfDataStream : public FlightDataStream {
   std::shared_ptr<Schema> schema() override { return schema_; }
 
   Status GetSchemaPayload(FlightPayload* payload) override {
-    return ipc::internal::GetSchemaPayload(*schema_, &dictionary_memo_,
+    return ipc::internal::GetSchemaPayload(*schema_, ipc_options_, &dictionary_memo_,
                                            &payload->ipc_message);
   }
 
@@ -103,8 +103,8 @@ class PerfDataStream : public FlightDataStream {
     } else {
       records_sent_ += batch_length_;
     }
-    return ipc::internal::GetRecordBatchPayload(*batch, default_memory_pool(),
-                                                &payload->ipc_message);
+    return ipc::internal::GetRecordBatchPayload(
+        *batch, ipc_options_, default_memory_pool(), &payload->ipc_message);
   }
 
  private:
@@ -115,6 +115,7 @@ class PerfDataStream : public FlightDataStream {
   int64_t records_sent_;
   std::shared_ptr<Schema> schema_;
   ipc::DictionaryMemo dictionary_memo_;
+  ipc::IpcOptions ipc_options_;
   std::shared_ptr<RecordBatch> batch_;
   ArrayVector arrays_;
 };

cpp/src/arrow/flight/server.cc

@@ -36,6 +36,7 @@
 
 #include "arrow/buffer.h"
 #include "arrow/ipc/dictionary.h"
+#include "arrow/ipc/options.h"
 #include "arrow/ipc/reader.h"
 #include "arrow/ipc/writer.h"
 #include "arrow/memory_pool.h"
@@ -640,13 +641,13 @@ class RecordBatchStream::RecordBatchStreamImpl {
 
   RecordBatchStreamImpl(const std::shared_ptr<RecordBatchReader>& reader,
                         MemoryPool* pool)
-      : pool_(pool), reader_(reader) {}
+      : pool_(pool), reader_(reader), ipc_options_(ipc::IpcOptions::Defaults()) {}
 
   std::shared_ptr<Schema> schema() { return reader_->schema(); }
 
   Status GetSchemaPayload(FlightPayload* payload) {
-    return ipc::internal::GetSchemaPayload(*reader_->schema(), &dictionary_memo_,
-                                           &payload->ipc_message);
+    return ipc::internal::GetSchemaPayload(*reader_->schema(), ipc_options_,
+                                           &dictionary_memo_, &payload->ipc_message);
   }
 
   Status Next(FlightPayload* payload) {
@@ -664,7 +665,7 @@ class RecordBatchStream::RecordBatchStreamImpl {
     if (stage_ == Stage::DICTIONARY) {
       if (dictionary_index_ == static_cast<int>(dictionaries_.size())) {
         stage_ = Stage::RECORD_BATCH;
-        return ipc::internal::GetRecordBatchPayload(*current_batch_, pool_,
+        return ipc::internal::GetRecordBatchPayload(*current_batch_, ipc_options_, pool_,
                                                     &payload->ipc_message);
       } else {
         return GetNextDictionary(payload);
@@ -679,15 +680,15 @@ class RecordBatchStream::RecordBatchStreamImpl {
       payload->ipc_message.metadata = nullptr;
       return Status::OK();
     } else {
-      return ipc::internal::GetRecordBatchPayload(*current_batch_, pool_,
+      return ipc::internal::GetRecordBatchPayload(*current_batch_, ipc_options_, pool_,
                                                   &payload->ipc_message);
     }
   }
 
  private:
   Status GetNextDictionary(FlightPayload* payload) {
     const auto& it = dictionaries_[dictionary_index_++];
-    return ipc::internal::GetDictionaryPayload(it.first, it.second, pool_,
+    return ipc::internal::GetDictionaryPayload(it.first, it.second, ipc_options_, pool_,
                                                &payload->ipc_message);
   }
 
@@ -703,6 +704,7 @@ class RecordBatchStream::RecordBatchStreamImpl {
   MemoryPool* pool_;
   std::shared_ptr<RecordBatchReader> reader_;
   ipc::DictionaryMemo dictionary_memo_;
+  ipc::IpcOptions ipc_options_;
   std::shared_ptr<RecordBatch> current_batch_;
   std::vector<std::pair<int64_t, std::shared_ptr<Array>>> dictionaries_;
 

cpp/src/arrow/ipc/message.h

@@ -24,6 +24,7 @@
 #include <memory>
 #include <string>
 
+#include "arrow/ipc/options.h"
 #include "arrow/status.h"
 #include "arrow/util/macros.h"
 #include "arrow/util/visibility.h"
@@ -57,11 +58,6 @@ enum class MetadataVersion : char {
   V4
 };
 
-// ARROW-109: We set this number arbitrarily to help catch user mistakes. For
-// deeply nested schemas, it is expected the user will indicate explicitly the
-// maximum allowed recursion depth
-constexpr int kMaxNestingDepth = 64;
-
 // Read interface classes. We do not fully deserialize the flatbuffers so that
 // individual fields metadata can be retrieved from very large schema without
 //

cpp/src/arrow/ipc/options.cc

@@ -0,0 +1,26 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/ipc/options.h"
+
+namespace arrow {
+namespace ipc {
+
+IpcOptions IpcOptions::Defaults() { return IpcOptions(); }
+
+}  // namespace ipc
+}  // namespace arrow

cpp/src/arrow/ipc/options.h

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstdint>
+
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+namespace ipc {
+
+// ARROW-109: We set this number arbitrarily to help catch user mistakes. For
+// deeply nested schemas, it is expected the user will indicate explicitly the
+// maximum allowed recursion depth
+constexpr int kMaxNestingDepth = 64;
+
+struct ARROW_EXPORT IpcOptions {
+  // If true, allow field lengths that don't fit in a signed 32-bit int.
+  // Some implementations may not be able to parse such streams.
+  bool allow_64bit = false;
+  // The maximum permitted schema nesting depth.
+  int max_recursion_depth = kMaxNestingDepth;
+
+  static IpcOptions Defaults();
+};
+
+}  // namespace ipc
+}  // namespace arrow

cpp/src/arrow/ipc/read-write-benchmark.cc

@@ -50,6 +50,7 @@ std::shared_ptr<RecordBatch> MakeRecordBatch(int64_t total_size, int64_t num_fie
 static void WriteRecordBatch(benchmark::State& state) {  // NOLINT non-const reference
   // 1MB
   constexpr int64_t kTotalSize = 1 << 20;
+  auto options = ipc::IpcOptions::Defaults();
 
   std::shared_ptr<ResizableBuffer> buffer;
   ABORT_NOT_OK(AllocateResizableBuffer(kTotalSize & 2, &buffer));
@@ -60,7 +61,7 @@ static void WriteRecordBatch(benchmark::State& state) {  // NOLINT non-const ref
     int32_t metadata_length;
     int64_t body_length;
     if (!ipc::WriteRecordBatch(*record_batch, 0, &stream, &metadata_length, &body_length,
-                               default_memory_pool())
+                               options, default_memory_pool())
              .ok()) {
       state.SkipWithError("Failed to write!");
     }
@@ -71,6 +72,7 @@ static void WriteRecordBatch(benchmark::State& state) {  // NOLINT non-const ref
 static void ReadRecordBatch(benchmark::State& state) {  // NOLINT non-const reference
   // 1MB
   constexpr int64_t kTotalSize = 1 << 20;
+  auto options = ipc::IpcOptions::Defaults();
 
   std::shared_ptr<ResizableBuffer> buffer;
   ABORT_NOT_OK(AllocateResizableBuffer(kTotalSize & 2, &buffer));
@@ -81,7 +83,7 @@ static void ReadRecordBatch(benchmark::State& state) {  // NOLINT non-const refe
   int32_t metadata_length;
   int64_t body_length;
   if (!ipc::WriteRecordBatch(*record_batch, 0, &stream, &metadata_length, &body_length,
-                             default_memory_pool())
+                             options, default_memory_pool())
            .ok()) {
     state.SkipWithError("Failed to write!");
   }