GH-45948: [C++][Parquet] Variant shredding

Implements variant shredding infrastructure per the VariantShredding spec,
achieving functional parity with arrow-rs parquet-variant-compute for the
primitive shredding path.

== VariantBuilder extensions ==

- BuildWithoutMeta(): raw value bytes without metadata (shared metadata)
- UnsafeAppendEncoded(data, size): zero-copy append of pre-encoded bytes
- SetAllowDuplicates(bool): last-value-wins duplicate key compaction
- FinishObject() updated with deterministic duplicate resolution

== VariantExtensionType evolution ==

- Supports shredded storage: struct{metadata, value?, typed_value?}
- IsSupportedStorageType() accepts both unshredded (2 required fields)
  and shredded (required metadata + optional value + optional typed_value)
- Added typed_value() accessor and is_shredded() query
- Constructor finds fields by name (not position) for robustness

== Shredding schema + type compatibility ==

- VariantShreddingSchema: Primitive/Object/Array tree schema
- ToArrowType(): converts to Arrow DataType with proper struct wrapping
- IsVariantCompatibleWithType(): 21-type compatibility matrix with
  widening rules (Int8->Int64, Float->Double, etc.)

== Shredding kernel (ShredVariantColumn) ==

- Per-row decode + type-match + routing to typed_value or residual value
- Compatible values routed to typed_value column (value set null)
- Incompatible values kept in value column (typed_value set null)
- Handles Variant Null as compatible with any target type

== Reconstruction kernel (ReconstructVariantColumn) ==

- Merges typed_value and residual value back into complete variant binary
- Handles all 4 states: (null,null)=missing, (v,null)=unshredded,
  (null,t)=shredded, (v,t)=partial object (errors for primitives)

== Tests ==

- 10 builder extension tests (BuildWithoutMeta, UnsafeAppendEncoded,
  SetAllowDuplicates with various scenarios)
- 20 type compatibility tests covering all primitive types
- 6 schema construction tests
- 4 shred/reconstruct round-trip tests proving identity:
  Reconstruct(Shred(v)) == v for matching, mixed, and null values

== Remaining work (object/array shredding, Parquet bridge) ==

- Object shredding: field-level routing + residual object encoding
- Array shredding: element-wise shredding with list builders
- Parquet bridge: VariantToNode/NodeToArrow in schema.cc
- Native typed_value extraction (currently stores variant bytes;
  full native-type extraction deferred to variant_get kernel)

Author: qzyu999

cpp/src/arrow/CMakeLists.txt

@@ -393,6 +393,7 @@ set(ARROW_SRCS
     extension/parquet_variant.cc
     extension/variant_builder.cc
     extension/variant_internal.cc
+    extension/variant_shredding.cc
     extension/uuid.cc
     pretty_print.cc
     record_batch.cc

cpp/src/arrow/extension/CMakeLists.txt

@@ -16,7 +16,8 @@
 # under the License.
 
 set(CANONICAL_EXTENSION_TESTS bool8_test.cc json_test.cc uuid_test.cc
-                              variant_internal_test.cc variant_builder_test.cc)
+                              variant_internal_test.cc variant_builder_test.cc
+                              variant_shredding_test.cc)
 
 if(ARROW_JSON)
   list(APPEND CANONICAL_EXTENSION_TESTS tensor_extension_array_test.cc opaque_test.cc)

cpp/src/arrow/extension/meson.build

@@ -16,7 +16,8 @@
 # under the License.
 
 canonical_extension_tests = ['bool8_test.cc', 'json_test.cc', 'uuid_test.cc',
-                             'variant_internal_test.cc', 'variant_builder_test.cc']
+                             'variant_internal_test.cc', 'variant_builder_test.cc',
+                             'variant_shredding_test.cc']
 
 if needs_json
     canonical_extension_tests += [
@@ -44,5 +45,6 @@ install_headers(
         # variant_internal.h: public API for variant binary encoding/decoding.
         # "internal" refers to the binary encoding internals, not visibility.
         'variant_internal.h',
+        'variant_shredding.h',
     ],
 )

cpp/src/arrow/extension/parquet_variant.cc

@@ -28,18 +28,20 @@ namespace arrow::extension {
 
 VariantExtensionType::VariantExtensionType(const std::shared_ptr<DataType>& storage_type)
     : ExtensionType(storage_type) {
-  // GH-45948: Shredded variants will need to handle an optional shredded_value as
-  // well as value_ becoming optional.
-
-  // IsSupportedStorageType should have been called already, asserting that both
-  // metadata and value are present.
-  if (storage_type->field(0)->name() == "metadata") {
-    metadata_ = storage_type->field(0);
-    value_ = storage_type->field(1);
-  } else {
-    value_ = storage_type->field(0);
-    metadata_ = storage_type->field(1);
+  // Find fields by name (ordering does not matter per spec).
+  for (int i = 0; i < storage_type->num_fields(); ++i) {
+    const auto& f = storage_type->field(i);
+    if (f->name() == "metadata") {
+      metadata_ = f;
+    } else if (f->name() == "value") {
+      value_ = f;
+    } else if (f->name() == "typed_value") {
+      typed_value_ = f;
+    }
   }
+  // IsSupportedStorageType() should have been called before construction.
+  DCHECK_NE(metadata_, nullptr);
+  DCHECK_NE(value_, nullptr);
 }
 
 bool VariantExtensionType::ExtensionEquals(const ExtensionType& other) const {
@@ -71,35 +73,52 @@ bool IsBinaryField(const std::shared_ptr<Field> field) {
 
 bool VariantExtensionType::IsSupportedStorageType(
     const std::shared_ptr<DataType>& storage_type) {
-  // For now we only supported unshredded variants. Unshredded variant storage
-  // type should be a struct with a binary metadata and binary value.
-  //
-  // GH-45948: In shredded variants, the binary value field can be replaced
-  // with one or more of the following: object, array, typed_value, and
-  // variant_value.
-  if (storage_type->id() == Type::STRUCT) {
-    if (storage_type->num_fields() == 2) {
-      // Ordering of metadata and value fields does not matter, as we will assign
-      // these to the VariantExtensionType's member shared_ptrs in the constructor.
-      // Here we just need to check that they are both present.
-
-      const auto& field0 = storage_type->field(0);
-      const auto& field1 = storage_type->field(1);
-
-      bool metadata_and_value_present =
-          (field0->name() == "metadata" && field1->name() == "value") ||
-          (field1->name() == "metadata" && field0->name() == "value");
-
-      if (metadata_and_value_present) {
-        // Both metadata and value must be non-nullable binary types for unshredded
-        // variants. This will change in GH-46948, when we will require a Visitor
-        // to traverse the structure of the variant.
-        return IsBinaryField(field0) && IsBinaryField(field1) && !field0->nullable() &&
-               !field1->nullable();
-      }
+  if (storage_type->id() != Type::STRUCT) {
+    return false;
+  }
+
+  // Find fields by name
+  std::shared_ptr<Field> metadata_field;
+  std::shared_ptr<Field> value_field;
+  std::shared_ptr<Field> typed_value_field;
+
+  for (int i = 0; i < storage_type->num_fields(); ++i) {
+    const auto& f = storage_type->field(i);
+    if (f->name() == "metadata") {
+      metadata_field = f;
+    } else if (f->name() == "value") {
+      value_field = f;
+    } else if (f->name() == "typed_value") {
+      typed_value_field = f;
     }
   }
 
+  // metadata is always required and must be binary-like
+  if (!metadata_field || !IsBinaryField(metadata_field)) {
+    return false;
+  }
+
+  // Unshredded: required metadata + required value (both binary)
+  if (value_field && !typed_value_field) {
+    return IsBinaryField(value_field) && !metadata_field->nullable() &&
+           !value_field->nullable();
+  }
+
+  // Shredded: required metadata + optional value + optional typed_value
+  if (value_field && typed_value_field) {
+    // metadata must be non-nullable, value must be nullable binary,
+    // typed_value must be nullable (any type)
+    return !metadata_field->nullable() && IsBinaryField(value_field) &&
+           value_field->nullable() && typed_value_field->nullable();
+  }
+
+  // NOTE: The shredding spec allows leaf schemas where `value` is absent
+  // (typed_value only, for fully-shredded columns with no residual). We
+  // reject this case for now because the current shredding implementation
+  // always produces a `value` column. Supporting value-absent schemas
+  // requires changes to ShredVariantColumn/ReconstructVariantColumn to
+  // handle the missing residual path. This can be added in a follow-up
+  // when Parquet reader integration requires it.
   return false;
 }
 
@@ -113,9 +132,12 @@ Result<std::shared_ptr<DataType>> VariantExtensionType::Make(
   return std::make_shared<VariantExtensionType>(std::move(storage_type));
 }
 
-/// NOTE: this is still experimental. GH-45948 will add shredding support, at which point
-/// we need to separate this into unshredded_variant and shredded_variant helper
-/// functions.
+/// \brief Return a VariantExtensionType instance.
+///
+/// Supports both unshredded and shredded storage types:
+/// - Unshredded: struct{required binary metadata, required binary value}
+/// - Shredded: struct{required binary metadata, optional binary value,
+///             optional <T> typed_value}
 std::shared_ptr<DataType> variant(std::shared_ptr<DataType> storage_type) {
   return VariantExtensionType::Make(std::move(storage_type)).ValueOrDie();
 }

cpp/src/arrow/extension/parquet_variant.h

@@ -40,6 +40,13 @@ class ARROW_EXPORT VariantArray : public ExtensionArray {
 ///   required binary value;
 /// }
 ///
+/// Shredded variant representation:
+/// optional group variant_name (VARIANT) {
+///   required binary metadata;
+///   optional binary value;
+///   optional <T> typed_value;
+/// }
+///
 /// To read more about variant encoding, see the variant encoding spec at
 /// https://github.com/apache/parquet-format/blob/master/VariantEncoding.md
 ///
@@ -69,10 +76,16 @@ class ARROW_EXPORT VariantExtensionType : public ExtensionType {
 
   std::shared_ptr<Field> value() const { return value_; }
 
+  /// \brief The typed_value field, or nullptr if unshredded.
+  std::shared_ptr<Field> typed_value() const { return typed_value_; }
+
+  /// \brief Whether this variant has a shredded typed_value column.
+  bool is_shredded() const { return typed_value_ != nullptr; }
+
  private:
-  // TODO GH-45948 added shredded_value
   std::shared_ptr<Field> metadata_;
-  std::shared_ptr<Field> value_;
+  std::shared_ptr<Field> value_;         // nullable when shredded
+  std::shared_ptr<Field> typed_value_;   // nullptr if unshredded
 };
 
 /// \brief Return a VariantExtensionType instance.

cpp/src/arrow/extension/variant_builder.cc

@@ -333,14 +333,34 @@ Status VariantBuilder::FinishObject(int64_t start, std::vector<FieldEntry>& fiel
     return Status::Invalid("VariantBuilder::FinishObject: invalid start position");
   }
 
-  // Sort fields by key name lexicographically (spec requirement)
-  std::sort(fields.begin(), fields.end(),
-            [](const FieldEntry& a, const FieldEntry& b) { return a.key < b.key; });
-
-  // Check for duplicate keys
-  for (size_t i = 1; i < fields.size(); ++i) {
-    if (fields[i].key == fields[i - 1].key) {
-      return Status::Invalid("VariantBuilder: duplicate key '", fields[i].key, "'");
+  // Sort fields by key name lexicographically (spec requirement).
+  // For allow_duplicates_ mode, use a secondary comparison on offset
+  // so that among duplicates, the last-inserted value (highest offset) is last.
+  std::sort(fields.begin(), fields.end(), [](const FieldEntry& a, const FieldEntry& b) {
+    if (a.key != b.key) return a.key < b.key;
+    return a.offset < b.offset;  // stable ordering for duplicates
+  });
+
+  // Handle duplicate keys
+  if (allow_duplicates_) {
+    // Last-value-wins compaction: among sorted duplicates (ascending offset),
+    // the last one has the highest offset = most recently inserted value.
+    // Reverse-iterate to collect unique keys keeping the last occurrence.
+    std::vector<FieldEntry> compacted;
+    compacted.reserve(fields.size());
+    for (auto it = fields.rbegin(); it != fields.rend(); ++it) {
+      if (compacted.empty() || compacted.back().key != it->key) {
+        compacted.push_back(std::move(*it));
+      }
+    }
+    std::reverse(compacted.begin(), compacted.end());
+    fields = std::move(compacted);
+  } else {
+    // Strict mode: reject duplicates
+    for (size_t i = 1; i < fields.size(); ++i) {
+      if (fields[i].key == fields[i - 1].key) {
+        return Status::Invalid("VariantBuilder: duplicate key '", fields[i].key, "'");
+      }
     }
   }
 
@@ -405,14 +425,12 @@ Result<VariantBuilder::EncodedVariant> VariantBuilder::Finish() {
     total_string_size += static_cast<int64_t>(k.size());
   }
 
-  // Validate sizes fit within the spec's 4-byte offset limit.
-  // Note: Go implementation enforces a stricter 128MB limit (metadataMaxSizeLimit).
-  // We only enforce the spec's 4-byte offset maximum (~4GB), which is the correct
-  // upper bound per the encoding format.
+  // Validate sizes fit within the spec's 4-byte offset limit (~4GB).
+  // This is the correct upper bound per the encoding format.
   if (total_string_size > static_cast<int64_t>(std::numeric_limits<uint32_t>::max())) {
-    return Status::Invalid(
-        "VariantBuilder: total dictionary string data (", total_string_size,
-        " bytes) exceeds maximum representable by 4-byte offsets");
+    return Status::Invalid("VariantBuilder: total dictionary string data (",
+                           total_string_size,
+                           " bytes) exceeds maximum representable by 4-byte offsets");
   }
 
   // Compute the offset_size: must accommodate both the largest string offset
@@ -472,4 +490,28 @@ Result<VariantBuilder::EncodedVariant> VariantBuilder::Finish() {
   return result;
 }
 
+Result<std::vector<uint8_t>> VariantBuilder::BuildWithoutMeta() {
+  if (buffer_.empty()) {
+    return Status::Invalid("VariantBuilder::BuildWithoutMeta: no value has been written");
+  }
+  std::vector<uint8_t> result = std::move(buffer_);
+  // After std::move, buffer_ is in a valid-but-unspecified state.
+  // Explicit clear() ensures a deterministic empty state for reuse.
+  buffer_.clear();
+  return result;
+}
+
+void VariantBuilder::UnsafeAppendEncoded(const uint8_t* data, int64_t size) {
+  // Callers must provide valid, non-empty variant bytes. The DCHECK guards
+  // catch programming errors in debug builds; in release builds, a zero-size
+  // append is a no-op (defensive against malformed input flowing through
+  // reconstruction paths).
+  DCHECK_NE(data, nullptr);
+  DCHECK_GT(size, 0);
+  if (size <= 0) return;
+  buffer_.insert(buffer_.end(), data, data + size);
+}
+
+void VariantBuilder::SetAllowDuplicates(bool allow) { allow_duplicates_ = allow; }
+
 }  // namespace arrow::extension::variant_internal