GH-32723: [C++][Parquet] Add option to use LARGE* variants of binary types

cpp/examples/parquet/parquet_arrow/reader_writer.cc

@@ -137,4 +137,4 @@ int main(int argc, char** argv) {
   read_single_rowgroup();
   read_single_column();
   read_single_column_chunk();
-}
+}

cpp/src/arrow/array/builder_dict.h

@@ -715,6 +715,29 @@ class Dictionary32Builder : public internal::DictionaryBuilderBase<Int32Builder,
   }
 };
 
+/// \brief A DictionaryArray builder that always returns int64 dictionary
+/// indices so that data cast to dictionary form will have a consistent index
+/// type, e.g. for creating a ChunkedArray
+template <typename T>
+class Dictionary64Builder : public internal::DictionaryBuilderBase<Int64Builder, T> {
+ public:
+  using BASE = internal::DictionaryBuilderBase<Int64Builder, T>;
+  using BASE::BASE;
+
+  /// \brief Append dictionary indices directly without modifying memo
+  ///
+  /// NOTE: Experimental API
+  Status AppendIndices(const int64_t* values, int64_t length,
+                       const uint8_t* valid_bytes = NULLPTR) {
+    int64_t null_count_before = this->indices_builder_.null_count();
+    ARROW_RETURN_NOT_OK(this->indices_builder_.AppendValues(values, length, valid_bytes));
+    this->capacity_ = this->indices_builder_.capacity();
+    this->length_ += length;
+    this->null_count_ += this->indices_builder_.null_count() - null_count_before;
+    return Status::OK();
+  }
+};
+
 // ----------------------------------------------------------------------
 // Binary / Unicode builders
 // (compatibility aliases; those used to be derived classes with additional
@@ -724,6 +747,7 @@ using BinaryDictionaryBuilder = DictionaryBuilder<BinaryType>;
 using StringDictionaryBuilder = DictionaryBuilder<StringType>;
 using BinaryDictionary32Builder = Dictionary32Builder<BinaryType>;
 using StringDictionary32Builder = Dictionary32Builder<StringType>;
+using BinaryDictionary64Builder = Dictionary64Builder<LargeBinaryType>;
 
 /// @}
 

cpp/src/arrow/type.h

@@ -678,6 +678,8 @@ class ARROW_EXPORT BaseBinaryType : public DataType {
 
 constexpr int64_t kBinaryMemoryLimit = std::numeric_limits<int32_t>::max() - 1;
 
+constexpr int64_t kLargeBinaryMemoryLimit = std::numeric_limits<int64_t>::max() - 1;
+
 /// \addtogroup binary-datatypes
 ///
 /// @{

cpp/src/parquet/arrow/reader.cc

@@ -219,6 +219,7 @@ class FileReaderImpl : public FileReader {
     ctx->iterator_factory = SomeRowGroupsFactory(row_groups);
     ctx->filter_leaves = true;
     ctx->included_leaves = included_leaves;
+    ctx->use_binary_large_variants = reader_properties_.use_binary_large_variants();
     return GetReader(manifest_.schema_fields[i], ctx, out);
   }
 
@@ -462,7 +463,7 @@ class LeafReader : public ColumnReaderImpl {
         input_(std::move(input)),
         descr_(input_->descr()) {
     record_reader_ = RecordReader::Make(
-        descr_, leaf_info, ctx_->pool, field_->type()->id() == ::arrow::Type::DICTIONARY);
+        descr_, leaf_info, ctx_->pool, field_->type()->id() == ::arrow::Type::DICTIONARY, false, ctx_->use_binary_large_variants);
     NextRowGroup();
   }
 

cpp/src/parquet/arrow/reader_internal.cc

@@ -85,6 +85,7 @@ using ::arrow::internal::SafeLeftShift;
 using ::arrow::util::SafeLoadAs;
 
 using parquet::internal::BinaryRecordReader;
+using parquet::internal::LargeBinaryRecordReader;
 using parquet::internal::DictionaryRecordReader;
 using parquet::internal::RecordReader;
 using parquet::schema::GroupNode;

cpp/src/parquet/arrow/reader_internal.h

@@ -109,6 +109,7 @@ struct ReaderContext {
   FileColumnIteratorFactory iterator_factory;
   bool filter_leaves;
   std::shared_ptr<std::unordered_set<int>> included_leaves;
+  bool use_binary_large_variants = false;
 
   bool IncludesLeaf(int leaf_index) const {
     if (this->filter_leaves) {

cpp/src/parquet/arrow/schema.cc

@@ -473,7 +473,7 @@ ::arrow::Result<std::shared_ptr<ArrowType>> GetTypeForNode(
     SchemaTreeContext* ctx) {
   ASSIGN_OR_RAISE(
       std::shared_ptr<ArrowType> storage_type,
-      GetArrowType(primitive_node, ctx->properties.coerce_int96_timestamp_unit()));
+      GetArrowType(primitive_node, ctx->properties.coerce_int96_timestamp_unit(), ctx->properties.use_binary_large_variants()));
   if (ctx->properties.read_dictionary(column_index) &&
       IsDictionaryReadSupported(*storage_type)) {
     return ::arrow::dictionary(::arrow::int32(), storage_type);

cpp/src/parquet/arrow/schema_internal.cc

@@ -127,6 +127,23 @@ Result<std::shared_ptr<ArrowType>> FromByteArray(const LogicalType& logical_type
   }
 }
 
+Result<std::shared_ptr<ArrowType>> FromLargeByteArray(const LogicalType& logical_type) {
+  switch (logical_type.type()) {
+    case LogicalType::Type::STRING:
+      return ::arrow::large_utf8();
+    case LogicalType::Type::DECIMAL:
+      return MakeArrowDecimal(logical_type);
+    case LogicalType::Type::NONE:
+    case LogicalType::Type::ENUM:
+    case LogicalType::Type::JSON:
+    case LogicalType::Type::BSON:
+      return ::arrow::large_binary();
+    default:
+      return Status::NotImplemented("Unhandled logical logical_type ",
+                                    logical_type.ToString(), " for binary array");
+  }
+}
+
 Result<std::shared_ptr<ArrowType>> FromFLBA(const LogicalType& logical_type,
                                             int32_t physical_length) {
   switch (logical_type.type()) {
@@ -181,7 +198,7 @@ Result<std::shared_ptr<ArrowType>> FromInt64(const LogicalType& logical_type) {
 
 Result<std::shared_ptr<ArrowType>> GetArrowType(
     Type::type physical_type, const LogicalType& logical_type, int type_length,
-    const ::arrow::TimeUnit::type int96_arrow_time_unit) {
+    const ::arrow::TimeUnit::type int96_arrow_time_unit, bool use_binary_large_variant) {
   if (logical_type.is_invalid() || logical_type.is_null()) {
     return ::arrow::null();
   }
@@ -200,7 +217,7 @@ Result<std::shared_ptr<ArrowType>> GetArrowType(
     case ParquetType::DOUBLE:
       return ::arrow::float64();
     case ParquetType::BYTE_ARRAY:
-      return FromByteArray(logical_type);
+      return use_binary_large_variant ? FromLargeByteArray(logical_type) : FromByteArray(logical_type);
     case ParquetType::FIXED_LEN_BYTE_ARRAY:
       return FromFLBA(logical_type, type_length);
     default: {
@@ -213,9 +230,10 @@ Result<std::shared_ptr<ArrowType>> GetArrowType(
 
 Result<std::shared_ptr<ArrowType>> GetArrowType(
     const schema::PrimitiveNode& primitive,
-    const ::arrow::TimeUnit::type int96_arrow_time_unit) {
+    const ::arrow::TimeUnit::type int96_arrow_time_unit,
+    bool use_binary_large_variant) {
   return GetArrowType(primitive.physical_type(), *primitive.logical_type(),
-                      primitive.type_length(), int96_arrow_time_unit);
+                      primitive.type_length(), int96_arrow_time_unit, use_binary_large_variant);
 }
 
 }  // namespace arrow

cpp/src/parquet/arrow/schema_internal.h

@@ -30,22 +30,27 @@ namespace arrow {
 using ::arrow::Result;
 
 Result<std::shared_ptr<::arrow::DataType>> FromByteArray(const LogicalType& logical_type);
+Result<std::shared_ptr<::arrow::DataType>> FromLargeByteArray(const LogicalType& logical_type);
+
 Result<std::shared_ptr<::arrow::DataType>> FromFLBA(const LogicalType& logical_type,
                                                     int32_t physical_length);
 Result<std::shared_ptr<::arrow::DataType>> FromInt32(const LogicalType& logical_type);
 Result<std::shared_ptr<::arrow::DataType>> FromInt64(const LogicalType& logical_type);
 
 Result<std::shared_ptr<::arrow::DataType>> GetArrowType(Type::type physical_type,
                                                         const LogicalType& logical_type,
-                                                        int type_length);
+                                                        int type_length,
+                                                        bool use_binary_large_variant = false);
 
 Result<std::shared_ptr<::arrow::DataType>> GetArrowType(
     Type::type physical_type, const LogicalType& logical_type, int type_length,
-    ::arrow::TimeUnit::type int96_arrow_time_unit = ::arrow::TimeUnit::NANO);
+    ::arrow::TimeUnit::type int96_arrow_time_unit = ::arrow::TimeUnit::NANO,
+    bool use_binary_large_variant = false);
 
 Result<std::shared_ptr<::arrow::DataType>> GetArrowType(
     const schema::PrimitiveNode& primitive,
-    ::arrow::TimeUnit::type int96_arrow_time_unit = ::arrow::TimeUnit::NANO);
+    ::arrow::TimeUnit::type int96_arrow_time_unit = ::arrow::TimeUnit::NANO,
+    bool use_binary_large_variant = false);
 
 }  // namespace arrow
 }  // namespace parquet