[DRAFT] HIVE-29287: Variant Shredding

Author: deniskuzZ

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveFileWriterFactory.java

@@ -19,6 +19,7 @@
 
 package org.apache.iceberg.mr.hive.writer;
 
+import java.util.Map;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortOrder;
@@ -31,9 +32,13 @@
 import org.apache.iceberg.data.parquet.GenericParquetWriter;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.types.Types;
 
 class HiveFileWriterFactory extends BaseFileWriterFactory<Record> {
 
+  private final Map<String, String> properties;
+  private Record sampleRecord = null;
+
   HiveFileWriterFactory(
       Table table,
       FileFormat dataFileFormat,
@@ -54,6 +59,7 @@ class HiveFileWriterFactory extends BaseFileWriterFactory<Record> {
         equalityDeleteRowSchema,
         equalityDeleteSortOrder,
         positionDeleteRowSchema);
+    properties = table.properties();
   }
 
   static Builder builderFor(Table table) {
@@ -78,6 +84,11 @@ protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
   @Override
   protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
     builder.createWriterFunc(GenericParquetWriter::create);
+    // Configure variant shredding function if conditions are met:
+    if (hasVariantColumns(dataSchema()) && isVariantShreddingEnabled(properties)) {
+      builder.variantShreddingFunc(
+          Parquet.constructVariantShreddingFunction(sampleRecord, dataSchema()));
+    }
   }
 
   @Override
@@ -149,4 +160,32 @@ HiveFileWriterFactory build() {
           positionDeleteRowSchema);
     }
   }
+
+  /**
+   * Check if the schema contains any variant columns.
+   */
+  private static boolean hasVariantColumns(Schema schema) {
+    return schema.columns().stream()
+        .anyMatch(field -> field.type() instanceof Types.VariantType);
+  }
+
+  /**
+   * Check if variant shredding is enabled via table properties.
+   */
+  private static boolean isVariantShreddingEnabled(Map<String, String> properties) {
+    String shreddingEnabled = properties.get("variant.shredding.enabled");
+    return "true".equalsIgnoreCase(shreddingEnabled);
+  }
+
+  /**
+   * Set a sample record to use for data-driven variant shredding schema generation.
+   * Should be called before the Parquet writer is created.
+   */
+  public void setSampleRecord(Record record) {
+    this.sampleRecord = record;
+  }
+
+  public Record getSampleRecord() {
+    return sampleRecord;
+  }
 }

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergRecordWriter.java

@@ -40,6 +40,8 @@ class HiveIcebergRecordWriter extends HiveIcebergWriterBase {
   private final Set<String> missingColumns;
   private final List<Types.NestedField> missingOrStructFields;
 
+  private final HiveFileWriterFactory fileWriterFactory;
+
   HiveIcebergRecordWriter(Table table, HiveFileWriterFactory fileWriterFactory,
       OutputFileFactory dataFileFactory, Context context) {
     super(table, newDataWriter(table, fileWriterFactory, dataFileFactory, context));
@@ -48,17 +50,20 @@ class HiveIcebergRecordWriter extends HiveIcebergWriterBase {
     this.missingColumns = context.missingColumns();
     this.missingOrStructFields = specs.get(currentSpecId).schema().asStruct().fields().stream()
         .filter(field -> missingColumns.contains(field.name()) || field.type().isStructType()).toList();
+    this.fileWriterFactory = fileWriterFactory;
   }
 
   @Override
   public void write(Writable row) throws IOException {
     Record record = ((Container<Record>) row).get();
     HiveSchemaUtil.setDefaultValues(record, missingOrStructFields, missingColumns);
 
+    if (fileWriterFactory.getSampleRecord() == null) {
+      fileWriterFactory.setSampleRecord(record);
+    }
     writer.write(record, specs.get(currentSpecId), partition(record, currentSpecId));
   }
 
-
   @Override
   public FilesForCommit files() {
     List<DataFile> dataFiles = ((DataWriteResult) writer.result()).dataFiles();