Support Iceberg v3 initial defaults in GPU reads [databricks]

iceberg-common/src/main/java/org/apache/iceberg/data/GpuFileHelpers.java

@@ -78,6 +78,17 @@ public static DeleteFile writeDeleteFile(
       throws IOException {
     int[] equalityFieldIds =
         deleteRowSchema.columns().stream().mapToInt(Types.NestedField::fieldId).toArray();
+    return writeDeleteFile(table, out, partition, deletes, deleteRowSchema, equalityFieldIds);
+  }
+
+  public static DeleteFile writeDeleteFile(
+      Table table,
+      OutputFile out,
+      StructLike partition,
+      List<Record> deletes,
+      Schema deleteRowSchema,
+      int[] equalityFieldIds)
+      throws IOException {
     FileWriterFactory<Record> factory =
         newWriterFactory(table, deleteRowSchema, equalityFieldIds);
 

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergShimUtils.java

@@ -39,6 +39,7 @@
 import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions;
 import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader;
 import org.apache.iceberg.spark.source.GpuSparkScan;
+import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.connector.read.Scan;
 import org.apache.spark.sql.connector.write.DeltaBatchWrite;
@@ -60,6 +61,16 @@ public interface IcebergShimUtils {
     /** Returns the Iceberg table format version from the table's current metadata. */
     int formatVersion(Table table);
 
+    /** Returns whether this field has an Iceberg v3 initial default. */
+    default boolean hasInitialDefault(Types.NestedField field) {
+        return false;
+    }
+
+    /** Converts an Iceberg v3 initial default to Spark's internal value representation. */
+    default Object initialDefaultToSpark(Types.NestedField field) {
+        throw new UnsupportedOperationException("Iceberg initial defaults are not supported");
+    }
+
     /**
      * Returns the fully qualified location URI of an Iceberg {@link ContentFile},
      * e.g. {@code "s3://bucket/path/to/file.parquet"} or {@code "file:/path/to/file.parquet"}.

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/ShimUtils.java

@@ -39,6 +39,7 @@
 import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions;
 import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader;
 import org.apache.iceberg.spark.source.GpuSparkScan;
+import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.connector.read.Scan;
 import org.apache.spark.sql.connector.write.DeltaBatchWrite;
@@ -78,6 +79,14 @@ public static int formatVersion(Table table) {
         return IMPL.formatVersion(table);
     }
 
+    public static boolean hasInitialDefault(Types.NestedField field) {
+        return IMPL.hasInitialDefault(field);
+    }
+
+    public static Object initialDefaultToSpark(Types.NestedField field) {
+        return IMPL.initialDefaultToSpark(field);
+    }
+
     public static int rowIdFieldId() {
         return ROW_ID_FIELD_ID;
     }

iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/IcebergFormatVersionSupport.scala

@@ -16,17 +16,37 @@
 
 package com.nvidia.spark.rapids.iceberg
 
+import scala.collection.JavaConverters._
+
 import com.nvidia.spark.rapids.{RapidsConf, RapidsMeta}
-import org.apache.iceberg.{Table, TableProperties}
+import org.apache.iceberg.{Schema, Table, TableProperties}
+import org.apache.iceberg.types.{Types, TypeUtil}
 
 import org.apache.spark.sql.execution.SparkPlan
 
 /** Common planning gate for Iceberg table format versions. */
 object IcebergFormatVersionSupport {
   private val MaxSupportedFormatVersion = 2
+  private val SupportedDefaultTypeIds = Set(
+    "BOOLEAN", "INTEGER", "LONG", "FLOAT", "DOUBLE", "DATE", "TIMESTAMP",
+    "STRING", "BINARY", "DECIMAL")
 
   def tagForFormatVersion(table: Table, meta: RapidsMeta[_, _, _]): Unit = {
-    tagForFormatVersion(ShimUtils.formatVersion(table), meta)
+    tagForFormatVersion(table, table.schema(), meta)
+  }
+
+  def tagForFormatVersion(
+      table: Table,
+      schemaToCheck: Schema,
+      meta: RapidsMeta[_, _, _]): Unit = {
+    val formatVersion = ShimUtils.formatVersion(table)
+    tagForFormatVersion(formatVersion, meta)
+    if (formatVersion > MaxSupportedFormatVersion && meta.conf.isIcebergV3Enabled) {
+      unsupportedDefault(schemaToCheck).foreach { case (path, typeName) =>
+        meta.willNotWorkOnGpu(
+          s"Iceberg default for field '$path' with type $typeName is not supported on GPU")
+      }
+    }
   }
 
   def tagForFormatVersion(
@@ -50,6 +70,64 @@ object IcebergFormatVersionSupport {
     }
   }
 
+  private[iceberg] def unsupportedDefault(schema: Schema): Option[(String, String)] = {
+    def find(fields: Seq[Types.NestedField], parent: String): Option[(String, String)] = {
+      fields.iterator.map { field =>
+        val path = if (parent.isEmpty) field.name() else s"$parent.${field.name()}"
+        val hasDefault = ShimUtils.hasInitialDefault(field)
+        val fieldType = field.`type`()
+
+        val unsupportedTimestampNtz = fieldType == Types.TimestampType.withoutZone()
+        if (hasDefault && fieldType.isPrimitiveType &&
+            (!SupportedDefaultTypeIds.contains(fieldType.typeId().name()) ||
+              unsupportedTimestampNtz)) {
+          Some(path -> fieldType.toString)
+        } else if (hasDefault && !fieldType.isPrimitiveType && !fieldType.isStructType) {
+          Some(path -> fieldType.toString)
+        } else if (fieldType.isNestedType) {
+          find(fieldType.asNestedType().fields().asScala.toSeq, path)
+        } else {
+          None
+        }
+      }.collectFirst { case Some(value) => value }
+    }
+
+    find(schema.columns().asScala.toSeq, "")
+  }
+
+  def withRequiredFields(
+      tableSchema: Schema,
+      expectedSchema: Schema,
+      requiredFieldIds: Seq[Int]): Schema = {
+    val projectedIds = TypeUtil.getProjectedIds(expectedSchema).asScala.map(_.toInt).toSet
+    val requiredFields = requiredFieldIds.distinct.map { fieldId =>
+      val field = Option(tableSchema.findField(fieldId)).getOrElse {
+        throw new IllegalArgumentException(s"Cannot find required field for ID $fieldId")
+      }
+
+      // Iceberg permits an equality-delete field to be a primitive nested in a struct. However,
+      // GpuDeleteFilter builds both its equality-delete schema and its probe expressions from the
+      // top-level columns in the table and required schemas. Adding a nested leaf as a top-level
+      // column here would make the planning check pass, but GpuDeleteFilter would later fail to
+      // resolve the field ID or bind it to the wrong input column. Validate every equality field,
+      // including fields already projected by the query, and fall back until GpuDeleteFilter can
+      // preserve and extract the complete nested path.
+      if (tableSchema.asStruct().field(fieldId) == null) {
+        throw new UnsupportedOperationException(
+          s"Nested equality-delete field '${field.name()}' with ID $fieldId is not supported " +
+            "on GPU")
+      }
+      field
+    }
+    val missingFields = requiredFields.filterNot(field => projectedIds.contains(field.fieldId()))
+
+    if (missingFields.isEmpty) {
+      expectedSchema
+    } else {
+      new Schema((expectedSchema.columns().asScala ++ missingFields).asJava)
+    }
+  }
+
   private def tagMergeRowsAncestor(
       meta: Option[RapidsMeta[_, _, _]],
       reason: String): Unit = meta match {

iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuParquetReaderPostProcessor.scala

@@ -138,6 +138,23 @@ private[iceberg] case class FetchConstant(
   }
 }
 
+/** Fill a field missing from an old data file with its Iceberg initial default. */
+private[iceberg] case class FillDefault(
+    fieldId: Int,
+    value: Any,
+    sparkType: DataType
+) extends ColumnAction {
+  override def execute(ctx: ColumnActionContext): CudfColumnVector = {
+    withResource(GpuScalar.from(value, sparkType)) { scalar =>
+      CudfColumnVector.fromScalar(scalar, ctx.numRows)
+    }
+  }
+
+  override def display(indent: Int): String = {
+    " " * indent + s"FillDefault(fieldId=$fieldId, ${sparkType.simpleString})"
+  }
+}
+
 /** Fill with null values for missing optional column. */
 private[iceberg] case class FillNull(sparkType: DataType) extends ColumnAction {
   override def execute(ctx: ColumnActionContext): CudfColumnVector = {
@@ -373,7 +390,6 @@ private[iceberg] case class ProcessStruct(
   }
 }
 
-
 /**
  * Helper object for building column actions when field is missing from file schema.
  * Shared between ActionBuildingVisitor and GpuParquetReaderPostProcessor.
@@ -384,10 +400,10 @@ private[iceberg] object MissingFieldActionBuilder {
    * Checks constants, metadata columns, and optionality.
    */
   def buildAction(
-      fieldId: Int,
+      field: Types.NestedField,
       sparkType: DataType,
-      isFieldOptional: Boolean,
       idToConstant: JMap[Integer, _]): ColumnAction = {
+    val fieldId = field.fieldId()
     // Row-lineage metadata needs per-row inheritance rather than ordinary constants.
     if (fieldId == ShimUtils.rowIdFieldId()) {
       return InheritRowId
@@ -412,11 +428,16 @@ private[iceberg] object MissingFieldActionBuilder {
       throw new UnsupportedOperationException("IS_DELETED meta column is not supported yet")
     }
 
-    // 3. Check if optional - fill null
-    if (isFieldOptional) {
+    // 3. Materialize Iceberg's initial default for data written before this field was added.
+    if (ShimUtils.hasInitialDefault(field)) {
+      return FillDefault(fieldId, ShimUtils.initialDefaultToSpark(field), sparkType)
+    }
+
+    // 4. Check if optional - fill null
+    if (field.isOptional) {
       FillNull(sparkType)
     } else {
-      // 4. Required field missing - throw error
+      // 5. Required field without an initial default is invalid.
       throw new IllegalArgumentException(s"Missing required field: $fieldId")
     }
   }
@@ -478,9 +499,8 @@ private class ActionBuildingVisitor(
         return FillNull(sparkType)
       }
       return MissingFieldActionBuilder.buildAction(
-        currentField.fieldId(),
+        currentField,
         sparkType,
-        currentField.isOptional,
         idToConstant)
     }
 
@@ -516,8 +536,7 @@ private class ActionBuildingVisitor(
   override def beforeField(field: Types.NestedField, partner: Type): Unit = {
     fieldStack.push((field, partner,
       isInsideConstantStruct ||
-        (field.`type`().isStructType &&
-          idToConstant.containsKey(field.fieldId()))))
+        (field.`type`().isStructType && idToConstant.containsKey(field.fieldId()))))
   }
 
   override def afterField(field: Types.NestedField, partner: Type): Unit = {
@@ -539,9 +558,8 @@ private class ActionBuildingVisitor(
         return FillNull(sparkType)
       }
       return MissingFieldActionBuilder.buildAction(
-        currentField.fieldId(),
+        currentField,
         sparkType,
-        currentField.isOptional,
         idToConstant)
     }
 
@@ -563,9 +581,8 @@ private class ActionBuildingVisitor(
         return FillNull(sparkType)
       }
       return MissingFieldActionBuilder.buildAction(
-        currentField.fieldId(),
+        currentField,
         sparkType,
-        currentField.isOptional,
         idToConstant)
     }
 
@@ -604,9 +621,8 @@ private class ActionBuildingVisitor(
       FillNull(expectedType)
     } else {
       MissingFieldActionBuilder.buildAction(
-        currentField.fieldId(),
+        currentField,
         expectedType,
-        currentField.isOptional,
         idToConstant)
     }
   }

iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/testutils/AddEqDeletes.scala

@@ -95,7 +95,8 @@ class AddEqDeletes extends UDF3[String, String, String, Unit] with Logging {
             outputFile,
             partitionValue.orNull,
             deleteRecords.map(_.asInstanceOf[Record]).asJava,
-            parquetFileInfo.deleteSchema)
+            parquetFileInfo.deleteSchema,
+            parquetFileInfo.equalityFieldIds.toArray)
         }.foldLeft(table.newRowDelta())(_.addDeletes(_))
         .commit()
 
@@ -107,9 +108,21 @@ class AddEqDeletes extends UDF3[String, String, String, Unit] with Logging {
 object AddEqDeletes extends Logging {
   private val PartitionColId = MetadataColumns.PARTITION_COLUMN_ID - 20000
 
+  private def primitiveFieldIds(field: NestedField): Seq[Int] = {
+    if (field.`type`().isPrimitiveType) {
+      Seq(field.fieldId())
+    } else if (field.`type`().isStructType) {
+      field.`type`().asStructType().fields().asScala.flatMap(primitiveFieldIds).toSeq
+    } else {
+      throw new IllegalArgumentException(
+        s"Equality-delete field ${field.name()} must be primitive or nested in a struct")
+    }
+  }
+
   case class ParquetFileInfo(
       fileSchema: ShadedMessageType,
       deleteSchema: Schema,
+      equalityFieldIds: Seq[Int],
       partitionSchema: Option[StructType]) {
     val readSchema: Schema = partitionSchema match {
       case Some(p) =>
@@ -127,13 +140,22 @@ object AddEqDeletes extends Logging {
       GenericParquetReaders.buildReader(readSchema, fileSchema)
 
     def nameMapping: NameMapping = {
+      def toMappedField(field: NestedField): MappedField = {
+        if (field.`type`().isStructType) {
+          val nestedFields = field.`type`().asStructType().fields().asScala.map(toMappedField)
+          MappedField.of(field.fieldId(), field.name(), MappedFields.of(nestedFields.toSeq: _*))
+        } else {
+          MappedField.of(field.fieldId(), field.name())
+        }
+      }
+
       val deleteFieldsMapping = deleteSchema
         .columns()
         .asScala
-        .map(f => MappedField.of(f.fieldId(), f.name()))
+        .map(toMappedField)
 
       val partitionFieldsMapping = partitionSchema
-        .map(_.fields().asScala.map(f => MappedField.of(f.fieldId(), f.name())))
+        .map(_.fields().asScala.map(toMappedField))
         .map(fields => MappedField.of(PartitionColId, MetadataColumns.PARTITION_COLUMN_NAME,
           MappedFields.of(fields.toSeq: _*)))
 
@@ -150,21 +172,24 @@ object AddEqDeletes extends Logging {
     withResource(icebergPartitionedFile.newReader()) { reader =>
       logDebug(s"Reading eq delete parquet file $parquetFile, " +
         s"schema:\n${reader.getFileMetaData.getSchema}")
-      val colNames = reader
+      val columnPaths = reader
         .getFileMetaData
         .getSchema
         .getColumns
         .asScala
-        .map(_.getPath.head)
-        .filterNot(MetadataColumns.isMetadataColumn)
+        .map(_.getPath)
+        .filterNot(path => MetadataColumns.isMetadataColumn(path.head))
+        .map(_.mkString("."))
         .toSeq
 
-      val deleteSchema = tableSchema.select(colNames: _*)
+      val deleteSchema = tableSchema.select(columnPaths: _*)
+      val equalityFieldIds = deleteSchema.columns().asScala.flatMap(primitiveFieldIds).toSeq
       val partitionSchema = Option(table.spec()).map(_.partitionType())
 
       ParquetFileInfo(reader.getFileMetaData.getSchema,
         deleteSchema,
+        equalityFieldIds,
         partitionSchema)
     }
   }
-}
+}