diff --git a/iceberg-common/src/main/java/org/apache/iceberg/data/GpuFileHelpers.java b/iceberg-common/src/main/java/org/apache/iceberg/data/GpuFileHelpers.java index 0780fd883ed..bcf889b96cb 100644 --- a/iceberg-common/src/main/java/org/apache/iceberg/data/GpuFileHelpers.java +++ b/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 deletes, + Schema deleteRowSchema, + int[] equalityFieldIds) + throws IOException { FileWriterFactory factory = newWriterFactory(table, deleteRowSchema, equalityFieldIds); diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergShimUtils.java b/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergShimUtils.java index 797cdffcfcf..b4c93ea9929 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergShimUtils.java +++ b/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergShimUtils.java @@ -20,6 +20,7 @@ import com.nvidia.spark.rapids.NoopMetric$; import com.nvidia.spark.rapids.RapidsConf; import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile; +import com.nvidia.spark.rapids.iceberg.parquet.ParquetReadUtils; import com.nvidia.spark.rapids.iceberg.parquet.converter.ParquetUnshader; import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile; import org.apache.hadoop.fs.Path; @@ -30,6 +31,7 @@ import org.apache.iceberg.Schema; import org.apache.iceberg.Table; import org.apache.iceberg.deletes.PositionDelete; +import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.io.DataWriteResult; import org.apache.iceberg.io.DeleteWriteResult; import org.apache.iceberg.io.FileIO; @@ -39,7 +41,9 @@ import org.apache.iceberg.parquet.GpuParquetIO; import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions; import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.iceberg.shaded.org.apache.parquet.schema.MessageType; 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; @@ -48,6 +52,7 @@ import java.io.IOException; import java.io.Serializable; import java.util.Map; +import java.util.function.IntPredicate; /** * Version-specific Iceberg API adapter. @@ -67,6 +72,40 @@ 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"); + } + + /** + * Prunes a Parquet schema while preserving the validity of default-only containers. + * + *

Runtimes before Iceberg 1.12 need the compatibility projection. A later shim can + * override this method when its native projection preserves the required physical leaves. + */ + default MessageType pruneColumns( + MessageType fileSchema, Schema requiredSchema, boolean useFallbackIds) { + return ParquetReadUtils.pruneColumns(fileSchema, requiredSchema, useFallbackIds); + } + + /** + * Creates a filter over row-group indexes, borrowing the reader for the predicate's lifetime. + * + *

Runtimes before Iceberg 1.12 need to retain groups when a filter references a missing + * field with an initial default. Later shims can use their default-aware native filters. + */ + default IntPredicate rowGroupFilter( + ParquetFileReader reader, Schema requiredSchema, MessageType fileSchema, + Expression filter, boolean caseSensitive) { + return ParquetReadUtils.rowGroupFilter( + reader, requiredSchema, fileSchema, filter, caseSensitive); + } + /** * 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"}. diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/ShimUtils.java b/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/ShimUtils.java index a16ccc91ddc..6f8d84d9916 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/ShimUtils.java +++ b/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/ShimUtils.java @@ -31,6 +31,7 @@ import org.apache.iceberg.Schema; import org.apache.iceberg.Table; import org.apache.iceberg.deletes.PositionDelete; +import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.io.DataWriteResult; import org.apache.iceberg.io.DeleteWriteResult; import org.apache.iceberg.io.FileIO; @@ -41,12 +42,14 @@ import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader; import org.apache.iceberg.shaded.org.apache.parquet.schema.MessageType; 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; import java.io.IOException; import java.util.Map; +import java.util.function.IntPredicate; /** * Dispatches to the correct version-specific Iceberg shim utilities based on the @@ -89,6 +92,25 @@ 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 MessageType pruneColumns( + MessageType fileSchema, Schema requiredSchema, boolean useFallbackIds) { + return IMPL.pruneColumns(fileSchema, requiredSchema, useFallbackIds); + } + + public static IntPredicate rowGroupFilter( + ParquetFileReader reader, Schema requiredSchema, MessageType fileSchema, + Expression filter, boolean caseSensitive) { + return IMPL.rowGroupFilter(reader, requiredSchema, fileSchema, filter, caseSensitive); + } + public static int rowIdFieldId() { return ROW_ID_FIELD_ID; } diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/IcebergFormatVersionSupport.scala b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/IcebergFormatVersionSupport.scala index da65ade1284..fddca8236dc 100644 --- a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/IcebergFormatVersionSupport.scala +++ b/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 { diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuParquetReaderPostProcessor.scala b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuParquetReaderPostProcessor.scala index 6f9f3c9fdf2..5c9965220a5 100644 --- a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuParquetReaderPostProcessor.scala +++ b/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) } } diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ParquetReadUtils.scala b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ParquetReadUtils.scala new file mode 100644 index 00000000000..7aa171b7644 --- /dev/null +++ b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ParquetReadUtils.scala @@ -0,0 +1,66 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. + * + * Licensed 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. + */ + +package com.nvidia.spark.rapids.iceberg.parquet + +import java.util.function.IntPredicate + +import org.apache.iceberg.Schema +import org.apache.iceberg.expressions.Expression +import org.apache.iceberg.parquet.{ParquetBloomRowGroupFilter, ParquetDictionaryRowGroupFilter, ParquetMetricsRowGroupFilter, ParquetSchemaUtil} +import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader +import org.apache.iceberg.shaded.org.apache.parquet.schema.MessageType + +/** Compatibility implementations used by Iceberg shims before 1.12. */ +object ParquetReadUtils { + def pruneColumns( + fileSchema: MessageType, + requiredSchema: Schema, + useFallbackIds: Boolean): MessageType = { + val projected = if (useFallbackIds) { + ParquetSchemaUtil.pruneColumnsFallback(fileSchema, requiredSchema) + } else { + ParquetSchemaUtil.pruneColumns(fileSchema, requiredSchema) + } + GpuIcebergParquetReader.preserveStructValidity(fileSchema, projected, requiredSchema) + } + + def rowGroupFilter( + reader: ParquetFileReader, + requiredSchema: Schema, + fileSchema: MessageType, + filter: Expression, + caseSensitive: Boolean): IntPredicate = { + // Older filters treat absent columns as null, ignoring initial defaults. Spark must + // evaluate these residuals after defaults are materialized, so retain every row group. + if (GpuIcebergParquetReader.hasMissingDefaultFilter( + fileSchema, requiredSchema, filter, caseSensitive)) { + (_: Int) => true + } else { + val stats = new ParquetMetricsRowGroupFilter(requiredSchema, filter, caseSensitive) + val dictionary = new ParquetDictionaryRowGroupFilter(requiredSchema, filter, caseSensitive) + val bloom = new ParquetBloomRowGroupFilter(requiredSchema, filter, caseSensitive) + val blocks = reader.getRowGroups + val shouldRead: IntPredicate = (index: Int) => { + val block = blocks.get(index) + stats.shouldRead(fileSchema, block) && + dictionary.shouldRead(fileSchema, block, reader.getDictionaryReader(block)) && + bloom.shouldRead(fileSchema, block, reader.getBloomFilterDataReader(block)) + } + shouldRead + } + } +} diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ParquetSchemaVisitor.scala b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ParquetSchemaVisitor.scala new file mode 100644 index 00000000000..3e4e38c02a9 --- /dev/null +++ b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ParquetSchemaVisitor.scala @@ -0,0 +1,62 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. + * + * Licensed 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. + */ + +package com.nvidia.spark.rapids.iceberg.parquet + +import scala.collection.JavaConverters._ +import scala.collection.mutable + +import org.apache.iceberg.shaded.org.apache.parquet.schema.{GroupType, PrimitiveType, Type} + +/** Visits physical fields, including the children of opaque logical groups such as VARIANT. */ +private[parquet] abstract class ParquetSchemaVisitor[T] { + def primitive(primitive: PrimitiveType): T + + def group(group: GroupType, children: Seq[T]): T + + /** Completes a field after its children, retaining its physical parent for wrapper handling. */ + def field(field: Type, parent: GroupType, result: T): T = result +} + +private[parquet] object ParquetSchemaVisitor { + // Iceberg's ParquetTypeVisitor gained a VARIANT callback after 1.9. Traverse the raw + // physical groups here so every supported runtime visits the same fields and IDs. + def visitFields[T](group: GroupType, visitor: ParquetSchemaVisitor[T]): Seq[T] = { + group.getFields.asScala.map { field => + val result = if (field.isPrimitive) { + visitor.primitive(field.asPrimitiveType()) + } else { + val childGroup = field.asGroupType() + visitor.group(childGroup, visitFields(childGroup, visitor)) + } + visitor.field(field, group, result) + }.toSeq + } + + def fieldIds(schema: GroupType): Set[Int] = { + val ids = mutable.HashSet.empty[Int] + visitFields(schema, new ParquetSchemaVisitor[Unit] { + override def primitive(primitive: PrimitiveType): Unit = () + + override def group(group: GroupType, children: Seq[Unit]): Unit = () + + override def field(field: Type, parent: GroupType, result: Unit): Unit = { + Option(field.getId).foreach(id => ids += id.intValue()) + } + }) + ids.toSet + } +} diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/reader.scala b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/reader.scala index ce0275cbf9d..77b308807fc 100644 --- a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/reader.scala +++ b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/reader.scala @@ -18,9 +18,10 @@ package com.nvidia.spark.rapids.iceberg.parquet import java.io.{IOException, UncheckedIOException} import java.net.URI -import java.util.Objects +import java.util.{Collections, Objects} import scala.collection.JavaConverters._ +import scala.collection.mutable import com.nvidia.spark.rapids.{CombineConf, DateTimeRebaseCorrected, GpuMetric, ThreadPoolConfBuilder} import com.nvidia.spark.rapids.Arm.withResource @@ -32,13 +33,20 @@ import com.nvidia.spark.rapids.shims.parquet.GpuParquetUtilsShims import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path import org.apache.iceberg.{MetadataColumns, Schema} -import org.apache.iceberg.expressions.Expression +import org.apache.iceberg.expressions.{Binder, Expression} import org.apache.iceberg.mapping.NameMapping import org.apache.iceberg.parquet._ import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader import org.apache.iceberg.shaded.org.apache.parquet.hadoop.metadata.{BlockMetaData => ShadedBlockMetaData} -import org.apache.iceberg.shaded.org.apache.parquet.schema.{MessageType => ShadedMessageType} +import org.apache.iceberg.shaded.org.apache.parquet.schema.{ + GroupType => ShadedGroupType, + MessageType => ShadedMessageType, + PrimitiveType => ShadedPrimitiveType, + Type => ShadedType, + Types => ShadedTypes +} +import org.apache.iceberg.shaded.org.apache.parquet.schema.LogicalTypeAnnotation.{ListLogicalTypeAnnotation, MapKeyValueTypeAnnotation, MapLogicalTypeAnnotation} import org.apache.parquet.hadoop.metadata.BlockMetaData import org.apache.spark.internal.Logging @@ -142,25 +150,210 @@ case class GpuIcebergParquetReaderConf( validateDeletionVectorCrc: Boolean, ) +object GpuIcebergParquetReader { + /** Decode helper leaves by physical type: only their definition/repetition levels are used. */ + private[iceberg] def physicalHelperSchema( + readSchema: ShadedMessageType, + requiredSchema: Schema): ShadedMessageType = { + val visitor = new PhysicalHelperVisitor(readSchema, requiredSchema) + new ShadedMessageType(readSchema.getName, + ParquetSchemaVisitor.visitFields(readSchema, visitor).asJava) + } + + private class PhysicalHelperVisitor(readSchema: ShadedMessageType, requiredSchema: Schema) + extends ParquetSchemaVisitor[ShadedType] { + private val usedIds = mutable.HashSet.empty[Int] + usedIds ++= requiredSchema.idToName().keySet().asScala.map(_.toInt) + usedIds ++= ParquetSchemaVisitor.fieldIds(readSchema) + private var nextHelperId = 1 + + override def primitive(primitive: ShadedPrimitiveType): ShadedType = { + if (primitive.getId != null && + requiredSchema.findField(primitive.getId.intValue()) != null) { + primitive + } else { + // An unprojected UUID, TIME or nanosecond timestamp can carry a parent's validity, + // although Spark's Parquet schema converter cannot decode its logical type. + val builder = ShadedTypes.primitive(primitive.getPrimitiveTypeName, primitive.getRepetition) + .length(primitive.getTypeLength) + Option(primitive.getId).foreach(id => builder.id(id.intValue())) + builder.named(primitive.getName) + } + } + + override def group(group: ShadedGroupType, children: Seq[ShadedType]): ShadedType = { + val annotation = group.getLogicalTypeAnnotation + val isCollection = annotation.isInstanceOf[ListLogicalTypeAnnotation] || + annotation.isInstanceOf[MapLogicalTypeAnnotation] || + annotation.isInstanceOf[MapKeyValueTypeAnnotation] + if (annotation != null && !isCollection && + (group.getId == null || requiredSchema.findField(group.getId.intValue()) == null)) { + // Opaque logical groups such as VARIANT are only needed for their validity here. + // Decode their retained children as an ordinary struct. Collection annotations must + // remain intact to preserve the physical LIST/MAP structure. + val builder = ShadedTypes.buildGroup(group.getRepetition) + children.foreach(child => builder.addField(child)) + Option(group.getId).foreach(id => builder.id(id.intValue())) + builder.named(group.getName) + } else { + group.withNewFields(children.asJava) + } + } + + override def field( + field: ShadedType, parent: ShadedGroupType, physical: ShadedType): ShadedType = { + val annotation = parent.getLogicalTypeAnnotation + val isWrapper = annotation.isInstanceOf[MapLogicalTypeAnnotation] || + (annotation.isInstanceOf[ListLogicalTypeAnnotation] && + !(ParquetSchemaUtil.determineListElementType(parent) eq parent.getType(0))) + if (physical.getId == null && !isWrapper) { + // convert() invents IDs starting at 1000 for unmapped fields. Such an ID can bind a + // helper to a real projected field, replacing its initial default with unrelated data. + while (usedIds.contains(nextHelperId)) { + nextHelperId += 1 + } + val helperId = nextHelperId + usedIds += helperId + physical.withId(helperId) + } else { + physical + } + } + } + + private[iceberg] def hasMissingDefaultFilter( + fileSchema: ShadedMessageType, + requiredSchema: Schema, + filter: Expression, + caseSensitive: Boolean): Boolean = { + val defaultFieldIds = Binder.boundReferences( + requiredSchema.asStruct(), Collections.singletonList(filter), caseSensitive) + .asScala.filter(id => ShimUtils.hasInitialDefault(requiredSchema.findField(id))) + + if (defaultFieldIds.isEmpty) { + false + } else { + val fileFieldIds = ParquetSchemaVisitor.fieldIds(fileSchema) + defaultFieldIds.exists(id => !fileFieldIds.contains(id.intValue())) + } + } + + /** Keep physical container paths when their projected children are missing from a file. */ + private[iceberg] def preserveStructValidity( + fileSchema: ShadedMessageType, + projectedSchema: ShadedMessageType, + requiredSchema: Schema): ShadedMessageType = { + def withChildren( + fileGroup: ShadedGroupType, + projectedGroup: ShadedGroupType, + children: Seq[ShadedType]): ShadedGroupType = { + val result = projectedGroup.withNewFields(children.asJava) + if (fileGroup.getLogicalTypeAnnotation.isInstanceOf[ListLogicalTypeAnnotation] && + !fileGroup.getType(0).isPrimitive && + (ParquetSchemaUtil.determineListElementType(fileGroup) eq fileGroup.getType(0)) && + !(ParquetSchemaUtil.determineListElementType(result) eq result.getType(0))) { + // A legacy two-level LIST can identify its repeated struct by its multiple fields. + // Keeping only one field would reinterpret that struct as a three-level list wrapper. + val fileElement = fileGroup.getType(0).asGroupType() + val element = result.getType(0).asGroupType() + val retained = element.getFields.asScala.map(f => f.getName -> f).toMap + val helper = fileElement.getFields.asScala.find(f => !retained.contains(f.getName)) + .map(firstLeaf(_)) + val fields = fileElement.getFields.asScala.flatMap { field => + retained.get(field.getName).orElse(helper.filter(_.getName == field.getName)) + } + result.withNewFields(Collections.singletonList[ShadedType]( + element.withNewFields(fields.asJava))) + } else { + result + } + } + + def firstLeaf(field: ShadedType, keepAllChildren: Boolean = false): ShadedType = { + if (field.isPrimitive) { + field + } else { + val group = field.asGroupType() + val children = group.getFields.asScala + // A map's repeated key/value group needs both children to remain a valid map schema. + val isMap = group.getLogicalTypeAnnotation.isInstanceOf[MapLogicalTypeAnnotation] + val retained = if (keepAllChildren) children else children.take(1) + withChildren(group, group, retained.map(child => firstLeaf(child, isMap)).toSeq) + } + } + + def isRequested(field: ShadedType): Boolean = { + if (field.getId != null) { + requiredSchema.findField(field.getId.intValue()) != null + } else { + // Parquet's repeated list and map wrapper groups do not have Iceberg field IDs. + !field.isPrimitive && field.asGroupType().getFields.asScala.exists(isRequested) + } + } + + def preserveChildren( + fileGroup: ShadedGroupType, + projectedGroup: ShadedGroupType): Seq[ShadedType] = { + val projectedByName = projectedGroup.getFields.asScala.map(f => f.getName -> f).toMap + fileGroup.getFields.asScala.flatMap { field => + projectedByName.get(field.getName) match { + case Some(projectedField) => Some(preserve(field, projectedField)) + case None if isRequested(field) => + // Restore every requested path before choosing helper leaves. One leaf under an + // ancestor cannot preserve a sibling collection's validity or element counts. + val emptyProjection = if (field.isPrimitive) { + field + } else { + field.asGroupType().withNewFields(Collections.emptyList[ShadedType]()) + } + Some(preserve(field, emptyProjection)) + case _ => None + } + }.toSeq + } + + def preserve(fileField: ShadedType, projectedField: ShadedType): ShadedType = { + if (projectedField.isPrimitive) { + projectedField + } else { + val projectedGroup = projectedField.asGroupType() + val children = preserveChildren(fileField.asGroupType(), projectedGroup) + if (children.isEmpty) { + // ParquetSchemaUtil.convert drops empty structs. A retained physical leaf carries + // definition levels for the parent even when the leaf itself is null. The + // post-processor removes this helper field while applying the projected defaults. + firstLeaf(fileField) + } else { + withChildren(fileField.asGroupType(), projectedGroup, children) + } + } + } + + new ShadedMessageType( + projectedSchema.getName, preserveChildren(fileSchema, projectedSchema).asJava) + } +} + trait GpuIcebergParquetReader extends Iterator[ColumnarBatch] with AutoCloseable with Logging { def conf: GpuIcebergParquetReaderConf def projectSchema(fileSchema: ShadedMessageType, requiredSchema: Schema): (ShadedMessageType, ShadedMessageType) = { - val (typeWithIds, fileReadSchema) = if (ParquetSchemaUtil.hasIds(fileSchema)) { - (fileSchema, ParquetSchemaUtil.pruneColumns(fileSchema, requiredSchema)) + val (typeWithIds, useFallbackIds) = if (ParquetSchemaUtil.hasIds(fileSchema)) { + (fileSchema, false) } else if (conf.nameMapping.isDefined) { val typeWithIds = ParquetSchemaUtil.applyNameMapping(fileSchema, conf.nameMapping.get) - (typeWithIds, ParquetSchemaUtil.pruneColumns(typeWithIds, requiredSchema)) + (typeWithIds, false) } else { - val typeWithIds = ParquetSchemaUtil.addFallbackIds(fileSchema) - (typeWithIds, ParquetSchemaUtil.pruneColumnsFallback(typeWithIds, requiredSchema)) + (ParquetSchemaUtil.addFallbackIds(fileSchema), true) } + val readSchemaWithValidity = + ShimUtils.pruneColumns(typeWithIds, requiredSchema, useFallbackIds) logDebug(s"Doing project schema, parquet file schema:\n$fileSchema " + s"\niceberg required schema:\n$requiredSchema " + - s"\nfile reade schema:\n$fileReadSchema") - (typeWithIds, fileReadSchema) + s"\nfile reade schema:\n$readSchemaWithValidity") + (typeWithIds, readSchemaWithValidity) } def filterRowGroups(reader: ParquetFileReader, @@ -172,21 +365,9 @@ trait GpuIcebergParquetReader extends Iterator[ColumnarBatch] with AutoCloseable .zipWithIndex filter.map { f => - val statsFilter = new ParquetMetricsRowGroupFilter(requiredSchema, - f, - conf.caseSensitive) - val dictFilter = new ParquetDictionaryRowGroupFilter(requiredSchema, - f, - conf.caseSensitive) - val bloomFilter = new ParquetBloomRowGroupFilter(requiredSchema, - f, - conf.caseSensitive) - - blocks.filter { case (rowGroup, _) => - statsFilter.shouldRead(typeWithIds, rowGroup) && - dictFilter.shouldRead(typeWithIds, rowGroup, reader.getDictionaryReader(rowGroup)) && - bloomFilter.shouldRead(typeWithIds, rowGroup, reader.getBloomFilterDataReader(rowGroup)) - } + val shouldRead = ShimUtils.rowGroupFilter( + reader, requiredSchema, typeWithIds, f, conf.caseSensitive) + blocks.filter { case (_, index) => shouldRead.test(index) } }.getOrElse(blocks) .toSeq } @@ -209,7 +390,7 @@ trait GpuIcebergParquetReader extends Iterator[ColumnarBatch] with AutoCloseable requiredSchema.findField(ShimUtils.rowIdFieldId()) != null } val initialProjection = projectSchema(fileSchema, requiredSchema) - val (typeWithIds, fileReadSchema) = + val (typeWithIds, projectedSchema) = // cuDF's deletion-vector Parquet path needs at least one physical data column. Force one // into metadata-only and count projections so native row-index filtering can determine // the surviving row count; the post-processor drops it from the requested output. @@ -222,6 +403,9 @@ trait GpuIcebergParquetReader extends Iterator[ColumnarBatch] with AutoCloseable } else { initialProjection } + // Use the original required schema so deletion-vector helpers are also decoded physically. + val fileReadSchema = + GpuIcebergParquetReader.physicalHelperSchema(projectedSchema, requiredSchema) val filteredBlocks = filterRowGroups(reader, requiredSchema, typeWithIds, file.filter) val blockFirstRowIndices: Seq[Long] = if (needsRowPosition || hasDeletionVector) { // `_pos` and inherited `_row_id` are file-global. When file.split is set the reader is diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/testutils/AddEqDeletes.scala b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/testutils/AddEqDeletes.scala index 855bb3b3fce..816cdd259e8 100644 --- a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/testutils/AddEqDeletes.scala +++ b/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) } } -} \ No newline at end of file +} diff --git a/iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkScan.scala b/iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkScan.scala index 9a62356ffea..61dd879335f 100644 --- a/iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkScan.scala +++ b/iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkScan.scala @@ -21,7 +21,7 @@ import scala.util.{Failure, Success, Try} import com.nvidia.spark.rapids._ import com.nvidia.spark.rapids.iceberg.{IcebergEncryptionSupport, IcebergFormatVersionSupport, ShimUtils} -import org.apache.iceberg.ScanTaskGroup +import org.apache.iceberg.{FileContent, ScanTask, ScanTaskGroup} import org.apache.iceberg.spark.GpuSparkReadConf import org.apache.iceberg.types.Types @@ -103,8 +103,28 @@ object GpuSparkScan { } Try { + val table = GpuSparkScanAccess.table(meta.wrapped) + val expectedSchema = GpuSparkScanAccess.expectedSchema(meta.wrapped) + val schemaToCheck = + if (ShimUtils.formatVersion(table) > 2 && meta.conf.isIcebergV3Enabled) { + val equalityDeleteFieldIds = GpuSparkScanAccess.taskGroups(meta.wrapped) + .asScala + .flatMap(_.tasks().asScala) + .map(_.asInstanceOf[ScanTask]) + .flatMap(_.asFileScanTask().deletes().asScala) + .filter(_.content() == FileContent.EQUALITY_DELETES) + .flatMap(_.equalityFieldIds().asScala) + .map(_.toInt) + .toSeq + IcebergFormatVersionSupport.withRequiredFields( + table.schema(), expectedSchema, equalityDeleteFieldIds) + } else { + expectedSchema + } IcebergFormatVersionSupport.tagForFormatVersion( - GpuSparkScanAccess.table(meta.wrapped), meta) + table, + schemaToCheck, + meta) } match { case Failure(e) => meta.willNotWorkOnGpu(s"error examining Iceberg table version: $e") case _ => diff --git a/iceberg/iceberg-1-10-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg110x/ShimUtilsImpl.java b/iceberg/iceberg-1-10-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg110x/ShimUtilsImpl.java index 7389d5e798d..6446aecce49 100644 --- a/iceberg/iceberg-1-10-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg110x/ShimUtilsImpl.java +++ b/iceberg/iceberg-1-10-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg110x/ShimUtilsImpl.java @@ -56,6 +56,16 @@ public int formatVersion(Table table) { return TableUtil.formatVersion(table); } + @Override + public boolean hasInitialDefault(Types.NestedField field) { + return field.initialDefaultLiteral() != null; + } + + @Override + public Object initialDefaultToSpark(Types.NestedField field) { + return SparkUtil.internalToSpark(field.type(), field.initialDefault()); + } + @Override public String locationOf(ContentFile f) { return f.location(); diff --git a/iceberg/iceberg-1-11-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg111x/ShimUtilsImpl.java b/iceberg/iceberg-1-11-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg111x/ShimUtilsImpl.java index 22fe9ac96c4..7f6740abaaf 100644 --- a/iceberg/iceberg-1-11-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg111x/ShimUtilsImpl.java +++ b/iceberg/iceberg-1-11-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg111x/ShimUtilsImpl.java @@ -56,6 +56,16 @@ public int formatVersion(Table table) { return TableUtil.formatVersion(table); } + @Override + public boolean hasInitialDefault(Types.NestedField field) { + return field.initialDefaultLiteral() != null; + } + + @Override + public Object initialDefaultToSpark(Types.NestedField field) { + return SparkUtil.internalToSpark(field.type(), field.initialDefault()); + } + @Override public String locationOf(ContentFile f) { return f.location(); diff --git a/iceberg/iceberg-1-9-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg19x/ShimUtilsImpl.java b/iceberg/iceberg-1-9-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg19x/ShimUtilsImpl.java index 67828ca42d3..7df217b188c 100644 --- a/iceberg/iceberg-1-9-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg19x/ShimUtilsImpl.java +++ b/iceberg/iceberg-1-9-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg19x/ShimUtilsImpl.java @@ -51,6 +51,16 @@ public int formatVersion(Table table) { return TableUtil.formatVersion(table); } + @Override + public boolean hasInitialDefault(Types.NestedField field) { + return field.initialDefaultLiteral() != null; + } + + @Override + public Object initialDefaultToSpark(Types.NestedField field) { + return SparkUtil.internalToSpark(field.type(), field.initialDefault()); + } + @Override public String locationOf(ContentFile f) { return f.location(); diff --git a/integration_tests/src/main/python/iceberg/__init__.py b/integration_tests/src/main/python/iceberg/__init__.py index 6ab2450ac8f..5ae5e7ad314 100644 --- a/integration_tests/src/main/python/iceberg/__init__.py +++ b/integration_tests/src/main/python/iceberg/__init__.py @@ -353,15 +353,20 @@ def _add_eq_deletes(spark: SparkSession, eq_delete_cols: List[str], row_count: i f"{eq_delete_col} can't be used as eq delete column" - spark_warehouse_dir = spark.conf.get("spark.sql.catalog.spark_catalog.warehouse") - - temp_dir = tempfile.mkdtemp(dir=spark_tmp_path) deletes = (spark.table(table_name) .select(eq_delete_cols + ["_partition"]) .distinct() .orderBy(eq_delete_cols) .limit(row_count) .repartition(1)) + + _add_eq_deletes_from_df(spark, deletes, table_name, spark_tmp_path) + + +def _add_eq_deletes_from_df(spark: SparkSession, deletes: DataFrame, table_name: str, + spark_tmp_path): + spark_warehouse_dir = spark.conf.get("spark.sql.catalog.spark_catalog.warehouse") + temp_dir = tempfile.mkdtemp(dir=spark_tmp_path) deletes.write.parquet(temp_dir, mode='overwrite') parquet_files = [f for f in os.listdir(temp_dir) if f.endswith(".parquet")] assert len(parquet_files) == 1, "Only one delete parquet file should be created" diff --git a/integration_tests/src/main/python/iceberg/iceberg_delete_test.py b/integration_tests/src/main/python/iceberg/iceberg_delete_test.py index 19d3765ab7b..be29c9fdc91 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_delete_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_delete_test.py @@ -13,13 +13,14 @@ # limitations under the License. import pytest -from pyspark.sql import functions as f +from pyspark.sql import Row, functions as f from asserts import (assert_equal_with_local_sort, assert_gpu_and_cpu_are_equal_collect, - assert_gpu_fallback_write_sql) -from conftest import is_iceberg_remote_catalog + assert_gpu_fallback_collect, assert_gpu_fallback_write_sql) +from conftest import is_iceberg_remote_catalog, spark_jvm from data_gen import * from iceberg import ( + _add_eq_deletes_from_df, iceberg_cow_format_versions, iceberg_mor_format_versions, with_iceberg_dml_session, with_iceberg_format_versions, create_iceberg_table, get_full_table_name, iceberg_write_enabled_conf, iceberg_base_table_cols, iceberg_gens_list, @@ -873,3 +874,282 @@ def read_func(spark, table_name): assert_gpu_fallback_write_sql( write_func, read_func, base_table_name, ["WriteDeltaExec"], conf=iceberg_delete_mor_enabled_conf) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Requires local equality-delete UDF") +@allow_non_gpu("BatchScanExec", "ColumnarToRowExec") +@ignore_order(local=True) +def test_iceberg_v3_default_on_implicit_equality_delete_field( + spark_tmp_table_factory, + spark_tmp_path, + register_iceberg_add_eq_deletes_udf): + table_name = get_full_table_name(spark_tmp_table_factory) + props = _build_tblprops({"format-version": "3"}) + props_sql = ", ".join(f"'{key}' = '{value}'" for key, value in props.items()) + initial_data_length = 2 + current_id = None + + def setup_table(spark): + nonlocal current_id + spark.sql( + f"CREATE TABLE {table_name} (id BIGINT) USING ICEBERG PARTITIONED BY (id) " + f"TBLPROPERTIES ({props_sql})") + initial_data = gen_df( + spark, + [("id", UniqueLongGen())], + length=initial_data_length).select(f.abs(f.col("id")).alias("id")) + target_id = initial_data.orderBy("id").first().id + initial_data.coalesce(1).writeTo(table_name).append() + + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + timestamp_ntz_type = jvm.org.apache.iceberg.types.Types.TimestampType.withoutZone() + default_value = jvm.org.apache.iceberg.expressions.Literal.of( + "2024-01-02T03:04:05").to(timestamp_ntz_type) + table.updateSchema().addColumn( + "_c9", + timestamp_ntz_type, + default_value).commit() + spark.sql(f"REFRESH TABLE {table_name}") + current_data = (gen_df( + spark, + [("id", UniqueLongGen()), + ("_c9", TimestampGen(nullable=False, tzinfo=None))], + length=1) + .select( + (f.abs(f.col("id")) + initial_data_length).alias("id"), + f.col("_c9"))) + current_id = current_data.first().id + current_data.coalesce(1).writeTo(table_name).append() + + # _c9 is intentionally omitted from the query below. The equality-delete file makes it an + # implicit required read field, and the old data file requires its initial default. + deletes = (spark.table(table_name) + .where(f.col("id") == target_id) + .select("_c9", "_partition") + .coalesce(1)) + _add_eq_deletes_from_df(spark, deletes, table_name, spark_tmp_path) + + with_cpu_session(setup_table) + remaining_rows = with_cpu_session( + lambda spark: spark.sql(f"SELECT id FROM {table_name} ORDER BY id").collect()) + assert len(remaining_rows) == initial_data_length + assert Row(current_id) in remaining_rows + assert_gpu_fallback_collect( + lambda spark: spark.sql(f"SELECT id FROM {table_name}"), + "BatchScanExec", + conf={"spark.rapids.sql.format.iceberg.v3.enabled": "true"}) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Requires local equality-delete UDF") +@allow_non_gpu("BatchScanExec", "ColumnarToRowExec") +@ignore_order(local=True) +def test_iceberg_v3_nested_equality_delete_field_falls_back( + spark_tmp_table_factory, + spark_tmp_path, + register_iceberg_add_eq_deletes_udf): + table_name = get_full_table_name(spark_tmp_table_factory) + props = _build_tblprops({"format-version": "3"}) + props_sql = ", ".join(f"'{key}' = '{value}'" for key, value in props.items()) + initial_data_length = 2 + current_data_length = 1 + current_ids = [] + + def setup_table(spark): + nonlocal current_ids + spark.sql( + f"CREATE TABLE {table_name} " + "(id BIGINT, payload STRUCT) USING ICEBERG PARTITIONED BY (id) " + f"TBLPROPERTIES ({props_sql})") + initial_data = gen_df( + spark, + [("id", UniqueLongGen()), + ("payload", StructGen( + [("existing", StringGen(nullable=False))], nullable=False))], + length=initial_data_length).select( + f.abs(f.col("id")).alias("id"), f.col("payload")) + target_id = initial_data.orderBy("id").first().id + initial_data.coalesce(1).writeTo(table_name).append() + + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + string_type = jvm.org.apache.iceberg.types.Types.StringType.get() + default_value = jvm.org.apache.iceberg.expressions.Literal.of("legacy") + table.updateSchema().addColumn( + "payload", "delete_key", string_type, default_value).commit() + spark.sql(f"REFRESH TABLE {table_name}") + current_data = (gen_df( + spark, + [("id", UniqueLongGen()), + ("payload", StructGen([ + ("existing", StringGen(nullable=False)), + ("delete_key", StringGen(pattern="current_[0-9]{4}", nullable=False)) + ], nullable=False))], + length=current_data_length) + .select( + (f.abs(f.col("id")) + initial_data_length).alias("id"), + f.col("payload"))) + current_ids = [row.id for row in current_data.select("id").collect()] + current_data.coalesce(1).writeTo(table_name).append() + + # Keep the parent struct in the Parquet delete row so Iceberg can preserve the nested field + # ID. Only delete_key is included below, making it the equality field rather than payload. + deletes = (spark.table(table_name) + .where(f.col("id") == target_id) + .select( + f.struct(f.col("payload.delete_key").alias("delete_key")).alias("payload"), + f.col("_partition")) + .coalesce(1)) + _add_eq_deletes_from_df(spark, deletes, table_name, spark_tmp_path) + + with_cpu_session(setup_table) + # Keep the nested equality field projected so the CPU baseline can apply the delete. The + # implicit-field fallback path is covered by GpuPostProcessorSuite. + read_query = f"SELECT id, payload.delete_key AS delete_key FROM {table_name}" + remaining_rows = with_cpu_session( + lambda spark: spark.sql(f"{read_query} ORDER BY id").collect()) + assert len(remaining_rows) == initial_data_length - 1 + current_data_length + remaining_ids = {row.id for row in remaining_rows} + assert all(current_id in remaining_ids for current_id in current_ids) + assert_gpu_fallback_collect( + lambda spark: spark.sql(read_query), + "BatchScanExec", + conf={"spark.rapids.sql.format.iceberg.v3.enabled": "true"}) + + +def _add_supported_delete_defaults(spark, table_name): + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + table.updateSchema().addColumn( + 'delete_key', jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(11)).addColumn( + 'string_key', jvm.org.apache.iceberg.types.Types.StringType.get(), + jvm.org.apache.iceberg.expressions.Literal.of('legacy')).commit() + spark.sql(f'REFRESH TABLE {table_name}') + + +def _assert_delete_rows(expected): + def check(cpu_rows, gpu_rows): + for rows in (cpu_rows, gpu_rows): + assert sorted([tuple(row) for row in rows], key=repr) == sorted(expected, key=repr) + return cpu_rows, gpu_rows + return check + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.skipif(is_iceberg_remote_catalog(), reason='Requires local equality-delete UDF') +@pytest.mark.parametrize('reader_type', ['PERFILE', 'MULTITHREADED']) +@pytest.mark.parametrize('project_keys', [False, True], ids=['implicit', 'projected']) +@ignore_order(local=True) +def test_iceberg_v3_supported_defaults_with_equality_deletes( + spark_tmp_table_factory, spark_tmp_path, register_iceberg_add_eq_deletes_udf, + reader_type, project_keys): + table_name = get_full_table_name(spark_tmp_table_factory) + + def setup(spark): + spark.sql(f'CREATE TABLE {table_name} (id BIGINT, p INT) USING ICEBERG PARTITIONED BY (p) ' + "TBLPROPERTIES ('format-version'='3')") + spark.createDataFrame([(0, 0), (1, 0), (2, 0)], 'id BIGINT, p INT') \ + .coalesce(1).writeTo(table_name).append() + _add_supported_delete_defaults(spark, table_name) + spark.createDataFrame([(3, 0, None, None), (4, 0, 11, 'legacy'), (5, 0, 23, 'current')], + 'id BIGINT, p INT, delete_key INT, string_key STRING') \ + .coalesce(1).writeTo(table_name).append() + # A multi-column delete distinguishes rows sharing the same default. Delete one row + # whose key is synthesized and one whose matching key is physically present. + deletes = spark.table(table_name).where('id IN (0, 4)') \ + .select('id', 'delete_key', 'string_key', '_partition').coalesce(1) + _add_eq_deletes_from_df(spark, deletes, table_name, spark_tmp_path) + files = spark.sql(f'SELECT content FROM {table_name}.delete_files').collect() + assert files and all(row.content == 2 for row in files) + + with_cpu_session(setup) + columns = 'id, delete_key, string_key' if project_keys else 'id' + expected = ([(1, 11, 'legacy'), (2, 11, 'legacy'), (3, None, None), (5, 23, 'current')] + if project_keys else [(1,), (2,), (3,), (5,)]) + assert_gpu_and_cpu_are_equal_collect( + lambda spark: spark.sql(f'SELECT {columns} FROM {table_name}'), + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=_assert_delete_rows(expected)) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.skipif(is_iceberg_remote_catalog(), reason='Requires local equality-delete UDF') +@pytest.mark.parametrize('reader_type', ['PERFILE', 'MULTITHREADED']) +@pytest.mark.parametrize('delete_kind', ['position', 'dv', 'dv-equality']) +@ignore_order(local=True) +def test_iceberg_v3_defaults_with_position_and_mixed_deletes( + spark_tmp_table_factory, spark_tmp_path, register_iceberg_add_eq_deletes_udf, + reader_type, delete_kind): + table_name = get_full_table_name(spark_tmp_table_factory) + + def setup(spark): + version = '2' if delete_kind == 'position' else '3' + spark.sql(f'CREATE TABLE {table_name} (id BIGINT, p INT) USING ICEBERG PARTITIONED BY (p) ' + f"TBLPROPERTIES ('format-version'='{version}', " + "'write.delete.mode'='merge-on-read')") + spark.createDataFrame([(i, 0) for i in range(6)], 'id BIGINT, p INT') \ + .coalesce(1).writeTo(table_name).append() + spark.sql(f'DELETE FROM {table_name} WHERE id = 0') + if delete_kind == 'position': + spark.sql(f"ALTER TABLE {table_name} SET TBLPROPERTIES ('format-version'='3')") + _add_supported_delete_defaults(spark, table_name) + spark.createDataFrame([(6, 0, None, None), (7, 0, 23, 'current')], + 'id BIGINT, p INT, delete_key INT, string_key STRING') \ + .coalesce(1).writeTo(table_name).append() + if delete_kind == 'dv-equality': + deletes = spark.table(table_name).where('id = 1') \ + .select('id', 'delete_key', 'string_key', '_partition').coalesce(1) + _add_eq_deletes_from_df(spark, deletes, table_name, spark_tmp_path) + files = {(row.content, row.file_format) for row in + spark.sql(f'SELECT content, file_format FROM {table_name}.delete_files').collect()} + expected = {(1, 'PARQUET')} if delete_kind == 'position' else {(1, 'PUFFIN')} + if delete_kind == 'dv-equality': + expected.add((2, 'PARQUET')) + assert files == expected + + with_cpu_session(setup) + legacy_ids = list(range(2 if delete_kind == 'dv-equality' else 1, 6)) + expected = [(i, 11, 'legacy') for i in legacy_ids] + [(6, None, None), (7, 23, 'current')] + conf = {'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type, + 'spark.sql.adaptive.enabled': 'false'} + + def read(spark, projection): + if spark.conf.get('spark.rapids.sql.enabled') != 'true': + # Iceberg 1.10.1 caches equality-delete rows in the first projection's field order. + # Keep the CPU oracle independent of that cache when changing projection order. + spark.conf.set('spark.sql.iceberg.executor-cache.delete-files.enabled', 'false') + return spark.sql(f'SELECT {projection} FROM {table_name}') + + # Old files have no projected physical column here. The DV reader must retain a physical + # helper, apply native filtering, and size generated defaults to the surviving row count. + assert_gpu_and_cpu_are_equal_collect( + lambda spark: read(spark, 'delete_key, string_key'), + conf=conf, + result_canonicalize_func_before_compare=_assert_delete_rows([row[1:] for row in expected])) + + def check_metadata(cpu_rows, gpu_rows): + for rows in (cpu_rows, gpu_rows): + assert sorted((row.id, row.delete_key, row.string_key) for row in rows) == expected + return cpu_rows, gpu_rows + + assert_gpu_and_cpu_are_equal_collect( + lambda spark: read( + spark, 'id, delete_key, string_key, _pos, _row_id, _last_updated_sequence_number'), + conf=conf, + result_canonicalize_func_before_compare=check_metadata) + assert_gpu_and_cpu_are_equal_collect( + lambda spark: read(spark, 'count(*)'), + conf=conf, + result_canonicalize_func_before_compare=_assert_delete_rows([(len(expected),)])) diff --git a/integration_tests/src/main/python/iceberg/iceberg_read_test.py b/integration_tests/src/main/python/iceberg/iceberg_read_test.py new file mode 100644 index 00000000000..d941a095184 --- /dev/null +++ b/integration_tests/src/main/python/iceberg/iceberg_read_test.py @@ -0,0 +1,949 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. +# +# Licensed 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. + +from decimal import Decimal +import json + +import pytest + +from asserts import assert_cpu_and_gpu_are_equal_collect_with_capture, \ + assert_gpu_and_cpu_are_equal_collect, assert_gpu_fallback_collect +from conftest import is_iceberg_remote_catalog, spark_jvm +from data_gen import LongGen, StructGen, UniqueLongGen, disable_parquet_field_id_write, gen_df +from iceberg import _build_tblprops, get_full_table_name, iceberg_unsupported_mark, \ + rapids_reader_types, runtime_iceberg_version, supports_iceberg_v3, ICEBERG_V3_UNSUPPORTED_REASON +from marks import allow_non_gpu, iceberg, ignore_order +from spark_session import is_spark_400_or_later, with_cpu_session + +pytestmark = iceberg_unsupported_mark + + +def _setup_iceberg_v3_defaults_table(table_name): + props = _build_tblprops({"format-version": "3"}) + props_sql = ", ".join(f"'{key}' = '{value}'" for key, value in props.items()) + + def setup_table(spark): + spark.sql( + f"CREATE TABLE {table_name} " + "(id BIGINT, s STRUCT) USING ICEBERG " + f"TBLPROPERTIES ({props_sql})") + gen_df( + spark, + [("id", UniqueLongGen()), + ("s", StructGen([("present", LongGen())]))], + length=32).coalesce(1).writeTo(table_name).append() + + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + literals = jvm.org.apache.iceberg.expressions.Literal + types = jvm.org.apache.iceberg.types.Types + update = table.updateSchema() + + def add_default(name, iceberg_type, value): + update.addColumn(name, iceberg_type, literals.of(value).to(iceberg_type)) + + # Cover every Iceberg 1.9 primitive type that Spark SQL can query. Iceberg TIME is omitted + # because Iceberg's Spark adapter rejects it with "Spark does not support time fields". + update.addRequiredColumn( + "required_added", types.IntegerType.get(), None, literals.of(7)) + update.addColumn("optional_added", types.StringType.get(), literals.of("legacy")) + update.addColumn("s", "nested_added", types.IntegerType.get(), literals.of(11)) + add_default("boolean_added", types.BooleanType.get(), True) + add_default("long_added", types.LongType.get(), jvm.java.lang.Long.valueOf(5000000000)) + add_default("float_added", types.FloatType.get(), jvm.java.lang.Float.valueOf("1.25")) + add_default("double_added", types.DoubleType.get(), 2.5) + add_default("date_added", types.DateType.get(), "2024-01-02") + add_default( + "timestamp_added", + types.TimestampType.withZone(), + "2024-01-02T03:04:05Z") + add_default( + "binary_added", types.BinaryType.get(), + jvm.java.nio.ByteBuffer.wrap(bytearray([1, 2, 3]))) + add_default( + "decimal_added", types.DecimalType.of(9, 2), + jvm.java.math.BigDecimal("12345.67")) + + # These types are representable by Iceberg/Spark but are not supported by the GPU + # default-materialization path and must cause scan fallback when projected. + add_default( + "timestamp_ntz_added", + types.TimestampType.withoutZone(), + "2024-01-02T03:04:05") + add_default( + "uuid_added", types.UUIDType.get(), + "123e4567-e89b-12d3-a456-426614174000") + add_default( + "fixed_added", types.FixedType.ofLength(3), + jvm.java.nio.ByteBuffer.wrap(bytearray([4, 5, 6]))) + update.commit() + spark.sql(f"REFRESH TABLE {table_name}") + + with_cpu_session(setup_table) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@ignore_order(local=True) +def test_iceberg_v3_initial_defaults_all_types(spark_tmp_table_factory): + table_name = get_full_table_name(spark_tmp_table_factory) + _setup_iceberg_v3_defaults_table(table_name) + v3_conf = {"spark.rapids.sql.format.iceberg.v3.enabled": "true"} + query = ( + f"SELECT id, s.present, s.nested_added, required_added, optional_added, " + "boolean_added, long_added, float_added, double_added, date_added, " + "timestamp_added, binary_added, decimal_added " + f"FROM {table_name} ORDER BY id") + assert_cpu_and_gpu_are_equal_collect_with_capture( + lambda spark: spark.sql(query), + exist_classes="GpuBatchScanExec", + conf=v3_conf) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@ignore_order(local=True) +def test_iceberg_v3_project_only_defaulted_struct_child(spark_tmp_table_factory, reader_type): + table_name = get_full_table_name(spark_tmp_table_factory) + props_sql = ", ".join( + f"'{key}' = '{value}'" for key, value in _build_tblprops({"format-version": "3"}).items()) + + def setup_table(spark): + spark.sql( + f"CREATE TABLE {table_name} " + "(id BIGINT, s STRUCT) USING ICEBERG " + f"TBLPROPERTIES ({props_sql})") + # A null physical leaf must still distinguish a non-null struct from a null parent. + spark.createDataFrame( + [(0, None), (1, (None,)), (2, (7,))], + "id BIGINT, s STRUCT" + ).coalesce(1).writeTo(table_name).append() + + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + table.updateSchema().addColumn( + "s", "nested_added", jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(11)).commit() + spark.sql(f"REFRESH TABLE {table_name}") + + with_cpu_session(setup_table) + + def check_defaults(cpu_rows, gpu_rows): + for rows in (cpu_rows, gpu_rows): + assert sorted((row.id, row.nested_added) for row in rows) == [ + (0, None), (1, 11), (2, 11)] + return cpu_rows, gpu_rows + + def read_defaults(spark): + # Iceberg's CPU nested pruning loses parent validity for this projection. Use an + # unpruned CPU reference while keeping GPU pruning enabled to exercise the regression. + gpu_enabled = spark.conf.get("spark.rapids.sql.enabled") == "true" + spark.conf.set("spark.sql.optimizer.nestedSchemaPruning.enabled", gpu_enabled) + return spark.sql(f"SELECT id, s.nested_added FROM {table_name}") + + assert_gpu_and_cpu_are_equal_collect( + read_defaults, + conf={ + "spark.rapids.sql.format.iceberg.v3.enabled": "true", + "spark.rapids.sql.format.parquet.reader.type": reader_type}, + result_canonicalize_func_before_compare=check_defaults) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@pytest.mark.parametrize('shape', ['list', 'map', 'multiple']) +@ignore_order(local=True) +def test_iceberg_v3_defaulted_collection_paths(spark_tmp_table_factory, reader_type, shape): + table_name = get_full_table_name(spark_tmp_table_factory) + old_element, new_element = 'STRUCT', 'STRUCT' + if shape == 'multiple': + old_fields = f'b ARRAY<{old_element}>, c MAP' + new_fields = f'b ARRAY<{new_element}>, c MAP' + old = [(0, None), (1, (1, None, {'x': (7,)})), (2, (2, [], None)), + (3, (3, [None], {})), + (4, (4, [(None,), (7,), None], {'a': (None,), 'b': None}))] + current = [(5, (5, [(9, 99), (None, None), None], + {'x': (8, 77), 'y': (None, None), 'z': None}))] + defaults = [('s.b.element', 11), ('s.c.value', 22)] + expressions = ['s.b.added', 'map_values(s.c).added'] + map_columns = [2] + expected = [(0, None, None), (1, None, [22]), (2, [], None), (3, [None], []), + (4, [11, 11, None], [22, None]), + (5, [99, None, None], [77, None, None])] + else: + is_map = shape == 'map' + old_fields = f'b MAP' if is_map else f'b ARRAY<{old_element}>' + new_fields = f'b MAP' if is_map else f'b ARRAY<{new_element}>' + values = [None, {}, {'a': None}, {'a': (None,), 'b': (7,), 'c': None}] if is_map else \ + [None, [], [None], [(None,), (7,), None]] + old = [(0, None)] + [(i, (i, value)) for i, value in enumerate(values, start=1)] + current_value = {'a': (9, 99), 'b': (None, None), 'c': None} if is_map else \ + [(9, 99), (None, None), None] + current = [(5, (5, current_value))] + defaults = [('s.b.value' if is_map else 's.b.element', 11)] + expressions = ['map_values(s.b).added' if is_map else 's.b.added'] + map_columns = [1] if is_map else [] + expected = [(0, None), (1, None), (2, []), (3, [None]), + (4, [11, 11, None]), (5, [99, None, None])] + + def setup(spark): + spark.sql(f'CREATE TABLE {table_name} ' + f'(id BIGINT, s STRUCT) USING ICEBERG ' + "TBLPROPERTIES ('format-version'='3')") + spark.createDataFrame(old, f'id BIGINT, s STRUCT') \ + .coalesce(1).writeTo(table_name).append() + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + update = table.updateSchema() + for parent, value in defaults: + update.addColumn(parent, 'added', jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(value)) + update.commit() + spark.sql(f'REFRESH TABLE {table_name}') + spark.createDataFrame(current, f'id BIGINT, s STRUCT') \ + .coalesce(1).writeTo(table_name).append() + + with_cpu_session(setup) + + def read(spark): + # Keep an unpruned CPU reference because Iceberg's CPU pruning also loses these paths. + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', + spark.conf.get('spark.rapids.sql.enabled') == 'true') + projections = ', '.join(f'{expr} AS added_{i}' for i, expr in enumerate(expressions)) + return spark.sql(f'SELECT id, {projections} FROM {table_name}') + + def normalize(rows): + return [tuple(sorted(value, key=repr) if i in map_columns and value is not None else value + for i, value in enumerate(row)) for row in rows] + + def check(cpu_rows, gpu_rows): + normalized_cpu, normalized_gpu = normalize(cpu_rows), normalize(gpu_rows) + _check_default_rows(normalize(expected))(normalized_cpu, normalized_gpu) + return normalized_cpu, normalized_gpu + + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=check) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@allow_non_gpu("BatchScanExec", "ColumnarToRowExec") +@ignore_order(local=True) +def test_iceberg_v3_unsupported_initial_defaults_fallback(spark_tmp_table_factory): + table_name = get_full_table_name(spark_tmp_table_factory) + _setup_iceberg_v3_defaults_table(table_name) + v3_conf = {"spark.rapids.sql.format.iceberg.v3.enabled": "true"} + + for unsupported_column in ["timestamp_ntz_added", "uuid_added", "fixed_added"]: + assert_gpu_fallback_collect( + lambda spark, column=unsupported_column: spark.sql( + f"SELECT id, {column} FROM {table_name}"), + "BatchScanExec", + conf=v3_conf) + + +def _check_default_rows(expected): + def check(cpu_rows, gpu_rows): + for rows in (cpu_rows, gpu_rows): + assert sorted([tuple(row) for row in rows], key=repr) == sorted(expected, key=repr) + return cpu_rows, gpu_rows + return check + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@pytest.mark.parametrize('project_default', [False, True], ids=['implicit', 'projected']) +@ignore_order(local=True) +def test_iceberg_v3_defaults_with_predicates( + spark_tmp_table_factory, reader_type, project_default): + table_name = get_full_table_name(spark_tmp_table_factory) + + def setup(spark): + spark.sql(f"CREATE TABLE {table_name} (id BIGINT) USING ICEBERG " + "TBLPROPERTIES ('format-version'='3')") + spark.createDataFrame([(0,), (1,)], 'id BIGINT').coalesce(1).writeTo(table_name).append() + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + table.updateSchema().addColumn( + 'added', jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(11)).commit() + spark.sql(f"REFRESH TABLE {table_name}") + spark.createDataFrame([(2, None), (3, 23)], 'id BIGINT, added INT') \ + .coalesce(1).writeTo(table_name).append() + frame = spark.table(table_name) + return frame.collect(), frame.schema + + reference_rows, reference_schema = with_cpu_session(setup) + assert sorted(tuple(row) for row in reference_rows) == [ + (0, 11), (1, 11), (2, None), (3, 23)] + predicates = [ + ('added = 11', [0, 1]), + ('added > 10', [0, 1, 3]), + ('added < 12', [0, 1]), + ('added IN (11, 23)', [0, 1, 3]), + ('added IS NULL', [2]), + ('added IS NOT NULL', [0, 1, 3]), + ('added != 23', [0, 1]), + ('added = 11 AND id = 1', [1]), + ('added = 11 OR id = 2', [0, 1, 2]), + ('NOT (added = 23 OR id = 0)', [1]), + ('added = 99', [])] + columns = ['id', 'added'] if project_default else ['id'] + for predicate, expected_ids in predicates: + def read(spark): + # Iceberg's CPU row-group pruning also treats an absent defaulted field as null. + # Materialize its unfiltered result before evaluating the CPU predicate. + frame = (spark.table(table_name) if + spark.conf.get('spark.rapids.sql.enabled') == 'true' else + spark.createDataFrame(reference_rows, reference_schema)) + return frame.where(predicate).select(*columns) + + expected = [tuple(row) if project_default else (row.id,) + for row in reference_rows if row.id in expected_ids] + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=_check_default_rows(expected)) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@pytest.mark.parametrize('shape', ['struct', 'list', 'map']) +@ignore_order(local=True) +def test_iceberg_v3_nested_defaults_mixed_files(spark_tmp_table_factory, reader_type, shape): + table_name = get_full_table_name(spark_tmp_table_factory) + if shape == 'struct': + old_type, new_type = 'STRUCT', 'STRUCT' + old = [(0, None), (1, (None,)), (2, (7,))] + current = [(4, None), (5, (None, None)), (6, (9, 99))] + parent, expression = 'c', 'c.added' + expected = [(0, None), (1, 11), (2, 11), (4, None), (5, None), (6, 99)] + elif shape == 'list': + old_type = 'ARRAY>' + new_type = 'ARRAY>' + old = [(0, None), (1, []), (2, [None]), (3, [(None,), (7,), None])] + current = [(4, None), (5, []), (6, [None, (None, None), (9, 99)])] + parent, expression = 'c.element', 'c.added' + expected = [(0, None), (1, []), (2, [None]), (3, [11, 11, None]), + (4, None), (5, []), (6, [None, None, 99])] + else: + old_type = 'MAP>' + new_type = 'MAP>' + old = [(0, None), (1, {}), (2, {'a': None}), + (3, {'a': (None,), 'b': (7,), 'c': None})] + current = [(4, None), (5, {}), (6, {'a': None, 'b': (None, None), 'c': (9, 99)})] + parent, expression = 'c.value', 'map_values(c).added' + expected = [(0, None), (1, []), (2, [None]), (3, [11, 11, None]), + (4, None), (5, []), (6, [None, None, 99])] + + expected = [(-1, None)] + expected + + def setup(spark): + spark.sql(f"CREATE TABLE {table_name} (id BIGINT) USING ICEBERG " + "TBLPROPERTIES ('format-version'='3', 'read.split.open-file-cost'='0')") + # This oldest file lacks the entire parent container: descendant defaults must not + # turn its null parent into a non-null struct, list or map. + spark.createDataFrame([(-1,)], 'id BIGINT').coalesce(1).writeTo(table_name).append() + spark.sql(f'ALTER TABLE {table_name} ADD COLUMN c {old_type}') + for rows in (old[:2], old[2:]): + spark.createDataFrame(rows, f'id BIGINT, c {old_type}') \ + .coalesce(1).writeTo(table_name).append() + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + table.updateSchema().addColumn( + parent, 'added', jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(11)).commit() + spark.sql(f"REFRESH TABLE {table_name}") + for rows in (current[:2], current[2:]): + spark.createDataFrame(rows, f'id BIGINT, c {new_type}') \ + .coalesce(1).writeTo(table_name).append() + + with_cpu_session(setup) + + def read(spark): + gpu = spark.conf.get('spark.rapids.sql.enabled') == 'true' + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', gpu) + return spark.sql(f'SELECT id, {expression} AS added FROM {table_name}') + + def check(cpu_rows, gpu_rows): + if shape != 'map': + return _check_default_rows(expected)(cpu_rows, gpu_rows) + # The order of map_values is unspecified; array-element order remains checked above. + def normalize(rows): + return [(row[0], None if row[1] is None else sorted(row[1], key=repr)) for row in rows] + normalized_expected = normalize(expected) + normalized_cpu, normalized_gpu = normalize(cpu_rows), normalize(gpu_rows) + _check_default_rows(normalized_expected)(normalized_cpu, normalized_gpu) + return normalized_cpu, normalized_gpu + + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=check) + + def collect_reference(spark): + frame = spark.table(table_name) + return frame.collect(), frame.schema + + reference_rows, reference_schema = with_cpu_session(collect_reference) + predicate = 'c.added = 11' if shape == 'struct' else f'array_contains({expression}, 11)' + expected_ids = [1, 2] if shape == 'struct' else [3] + filtered_expected = [row for row in expected if row[0] in expected_ids] + + def read_filtered(spark): + gpu = spark.conf.get('spark.rapids.sql.enabled') == 'true' + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', gpu) + frame = (spark.table(table_name) if gpu else + spark.createDataFrame(reference_rows, reference_schema)) + return frame.where(predicate).selectExpr('id', f'{expression} AS added') + + def check_filtered(cpu_rows, gpu_rows): + if shape == 'map': + cpu_rows = [(row[0], sorted(row[1], key=repr)) for row in cpu_rows] + gpu_rows = [(row[0], sorted(row[1], key=repr)) for row in gpu_rows] + wanted = [(row[0], sorted(row[1], key=repr)) for row in filtered_expected] + else: + wanted = filtered_expected + return _check_default_rows(wanted)(cpu_rows, gpu_rows) + + assert_gpu_and_cpu_are_equal_collect( + read_filtered, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=check_filtered) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@ignore_order(local=True) +def test_iceberg_v3_defaults_combining_and_type_evolution(spark_tmp_table_factory, reader_type): + table_name = get_full_table_name(spark_tmp_table_factory) + defaults = (Decimal('1.25'), Decimal('1234567890.25'), Decimal('12345678901234567890.25')) + physical = (Decimal('2.50'), Decimal('2345678901.25'), Decimal('-22345678901234567890.25')) + expected = [(11, i, *defaults) for i in range(6)] + [ + (None, 6, None, None, None), (7, 7, *physical), (5000000000, 8, *defaults)] + file_count = 0 + + def setup(spark): + nonlocal file_count + spark.sql(f"CREATE TABLE {table_name} (id BIGINT) USING ICEBERG TBLPROPERTIES (" + "'format-version'='3', 'read.split.open-file-cost'='0', " + "'read.split.target-size'='134217728')") + for i in range(6): + spark.createDataFrame([(i,)], 'id BIGINT').coalesce(1).writeTo(table_name).append() + jvm = spark_jvm() + types = jvm.org.apache.iceberg.types.Types + literals = jvm.org.apache.iceberg.expressions.Literal + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + update = table.updateSchema().addColumn('added', types.IntegerType.get(), literals.of(11)) + for name, precision, value in zip(('d32', 'd64', 'd128'), (9, 18, 38), defaults): + dtype = types.DecimalType.of(precision, 2) + update.addColumn(name, dtype, literals.of(jvm.java.math.BigDecimal(str(value)))) + update.commit() + spark.sql(f'REFRESH TABLE {table_name}') + schema = 'id BIGINT, added INT, d32 DECIMAL(9,2), d64 DECIMAL(18,2), d128 DECIMAL(38,2)' + for row in [(6, None, None, None, None), (7, 7, *physical)]: + spark.createDataFrame([row], schema).coalesce(1).writeTo(table_name).append() + table.refresh() + table.updateSchema().updateColumn('added', types.LongType.get()) \ + .moveFirst('added').renameColumn('added', 'renamed').commit() + spark.sql(f'REFRESH TABLE {table_name}') + spark.createDataFrame([(5000000000, 8, *defaults)], + 'renamed BIGINT, id BIGINT, d32 DECIMAL(9,2), d64 DECIMAL(18,2), d128 DECIMAL(38,2)') \ + .coalesce(1).writeTo(table_name).append() + table.refresh() + file_count = spark.sql(f'SELECT count(*) FROM {table_name}.data_files').first()[0] + assert file_count == 9 + groups = table.newScan().planTasks() + try: + iterator = groups.iterator() + assert iterator.hasNext() + assert iterator.next().files().size() == file_count + assert not iterator.hasNext() + finally: + groups.close() + + with_cpu_session(setup) + gpu_frame = [] + + def read(spark): + gpu = spark.conf.get('spark.rapids.sql.enabled') == 'true' + # Iceberg's CPU Arrow reader copies decimal defaults onto its physical int/long + # fields and fails to cast them. Use its row reader as the independent CPU oracle. + spark.conf.set('spark.sql.iceberg.vectorization.enabled', gpu) + frame = spark.sql(f'SELECT renamed, id, d32, d64, d128 FROM {table_name}') + if gpu: + gpu_frame.append(frame) + return frame + + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type, + 'spark.rapids.sql.reader.multithreaded.combine.sizeBytes': '64m', + 'spark.sql.adaptive.enabled': 'false'}, + result_canonicalize_func_before_compare=_check_default_rows(expected)) + if reader_type == 'COALESCING': + nodes = [gpu_frame[0]._jdf.queryExecution().executedPlan()] + while nodes: + node = nodes.pop() + if node.getClass().getSimpleName() == 'GpuBatchScanExec': + batches = node.metrics().get('numOutputBatches').get().value() + assert 0 < batches < file_count, \ + f'Expected actual combining, found {batches} batches' + break + children = node.children() + nodes.extend(children.apply(i) for i in range(children.size())) + else: + raise AssertionError('GpuBatchScanExec was not executed') + + def collect_reference(spark): + frame = read(spark) + return frame.collect(), frame.schema + + reference_rows, reference_schema = with_cpu_session(collect_reference) + + def read_filtered(spark): + gpu = spark.conf.get('spark.rapids.sql.enabled') == 'true' + spark.conf.set('spark.sql.iceberg.vectorization.enabled', gpu) + frame = (spark.table(table_name) if gpu else + spark.createDataFrame(reference_rows, reference_schema)) + return frame.where('renamed = 11 OR id = 7').select('renamed', 'id', 'd32', 'd64', 'd128') + + assert_gpu_and_cpu_are_equal_collect( + read_filtered, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=_check_default_rows( + [row for row in expected if row[1] < 6 or row[1] == 7])) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.skipif(is_iceberg_remote_catalog(), reason='add_files requires a local catalog') +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@ignore_order(local=True) +def test_iceberg_v3_nested_defaults_name_mapping(spark_tmp_table_factory, reader_type): + table_name = get_full_table_name(spark_tmp_table_factory) + source_name = get_full_table_name(spark_tmp_table_factory) + + def setup(spark): + spark.sql(f'CREATE TABLE {table_name} (unused INT, id BIGINT, s STRUCT) ' + "USING ICEBERG TBLPROPERTIES ('format-version'='3')") + spark.sql(f'ALTER TABLE {table_name} DROP COLUMN unused') + spark.sql(f'CREATE TABLE {source_name} ' + '(id BIGINT, s STRUCT) USING PARQUET') + spark.createDataFrame([(0, None), (1, (None,)), (2, (7,))], + 'id BIGINT, s STRUCT').coalesce(1).write.insertInto(source_name) + spark.sql(f"CALL spark_catalog.system.add_files(table => '{table_name}', " + f"source_table => '{source_name}')") + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + assert table.schema().findField('id').fieldId() != 1 + assert table.properties().get('schema.name-mapping.default') is not None + table.updateSchema().addColumn('s', 'added', + jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(11)) \ + .renameColumn('s.present', 'legacy').renameColumn('s', 'renamed').commit() + spark.sql(f'REFRESH TABLE {table_name}') + + with_cpu_session(setup, conf=disable_parquet_field_id_write) + + def read(spark): + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', + spark.conf.get('spark.rapids.sql.enabled') == 'true') + return spark.sql(f'SELECT id, renamed.added FROM {table_name}') + + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=_check_default_rows([(0, None), (1, 11), (2, 11)])) + + +@iceberg +@pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@ignore_order(local=True) +def test_iceberg_v3_default_with_unprojected_timestamp_ntz(spark_tmp_table_factory, reader_type): + table_name = get_full_table_name(spark_tmp_table_factory) + + def setup(spark): + spark.sql(f'CREATE TABLE {table_name} (id BIGINT, s STRUCT) ' + "USING ICEBERG TBLPROPERTIES ('format-version'='3')") + spark.sql(f"INSERT INTO {table_name} VALUES (0, NULL), (1, named_struct('legacy', " + "CAST(NULL AS TIMESTAMP_NTZ))), (2, named_struct('legacy', " + "CAST('2024-01-02 03:04:05' AS TIMESTAMP_NTZ)))") + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + table.updateSchema().addColumn('s', 'added', + jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(11)).commit() + spark.sql(f'REFRESH TABLE {table_name}') + + with_cpu_session(setup) + + def read(spark): + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', + spark.conf.get('spark.rapids.sql.enabled') == 'true') + return spark.sql(f'SELECT id, s.added FROM {table_name}') + + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=_check_default_rows([(0, None), (1, 11), (2, 11)])) + + +@iceberg +@pytest.mark.parametrize('format_version', ['2', pytest.param('3', marks=pytest.mark.skipif( + not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON))]) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@pytest.mark.parametrize('shape', ['struct', 'list', 'map']) +@ignore_order(local=True) +def test_iceberg_missing_nested_field_with_unprojected_uuid( + spark_tmp_table_factory, format_version, reader_type, shape): + table_name = get_full_table_name(spark_tmp_table_factory) + default = 11 if format_version == '3' else None + nested_type = {'type': 'struct', 'fields': [ + {'id': 3, 'name': 'legacy', 'required': False, 'type': 'uuid'}]} + if shape == 'list': + field_type = {'type': 'list', 'element-id': 4, + 'element-required': False, 'element': nested_type} + parent, expression = 'c.element', 'c.added' + expected = [(0, None), (1, []), (2, [None]), (3, [default, default, None])] + elif shape == 'map': + field_type = {'type': 'map', 'key-id': 4, 'key': 'string', + 'value-id': 5, 'value-required': False, 'value': nested_type} + parent, expression = 'c.value', 'map_values(c).added' + expected = [(0, None), (1, []), (2, [None]), (3, [default, default, None])] + else: + field_type = nested_type + parent, expression = 'c', 'c.added' + expected = [(0, None), (1, default), (2, default)] + + def setup(spark): + jvm = spark_jvm() + schema = jvm.org.apache.iceberg.SchemaParser.fromJson(json.dumps({ + 'type': 'struct', 'fields': [ + {'id': 1, 'name': 'id', 'required': True, 'type': 'int'}, + {'id': 2, 'name': 'c', 'required': False, 'type': field_type}]})) + catalog = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergCatalog( + spark._jsparkSession, 'spark_catalog') + props = jvm.java.util.HashMap() + for key, value in _build_tblprops({'format-version': format_version}).items(): + props.put(key, value) + table = catalog.createTable(jvm.org.apache.iceberg.catalog.TableIdentifier.parse(table_name), + schema, jvm.org.apache.iceberg.PartitionSpec.unpartitioned(), props) + schema = table.schema() + records = jvm.org.apache.iceberg.data.GenericRecord + old_struct = schema.findType('c') + if shape == 'list': + old_struct = old_struct.asListType().elementType() + elif shape == 'map': + old_struct = old_struct.asMapType().valueType() + null_leaf = records.create(old_struct.asStructType()) + uuid_leaf = records.create(old_struct.asStructType()) + uuid_leaf.setField('legacy', jvm.java.util.UUID.fromString( + '123e4567-e89b-12d3-a456-426614174000')) + if shape == 'list': + empty, null_element, elements = [jvm.java.util.ArrayList() for _ in range(3)] + null_element.add(None) + for value in (null_leaf, uuid_leaf, None): + elements.add(value) + values = [None, empty, null_element, elements] + elif shape == 'map': + empty, null_element, elements = [jvm.java.util.LinkedHashMap() for _ in range(3)] + null_element.put('a', None) + for key, value in zip(('a', 'b', 'c'), (null_leaf, uuid_leaf, None)): + elements.put(key, value) + values = [None, empty, null_element, elements] + else: + values = [None, null_leaf, uuid_leaf] + + # Use Iceberg's writer because Spark SQL represents UUID as STRING and cannot create + # the original UUID-annotated physical column used as the validity helper. + path = table.location() + '/data/uuid-helper.parquet' + appender = jvm.org.apache.iceberg.data.GenericAppenderFactory(schema).newAppender( + table.io().newOutputFile(path), jvm.org.apache.iceberg.FileFormat.PARQUET) + try: + for index, value in enumerate(values): + row = records.create(schema) + row.setField('id', index) + row.setField('c', value) + appender.add(row) + finally: + appender.close() + data_file = jvm.org.apache.iceberg.DataFiles.builder(table.spec()).withPath(path) \ + .withFormat(jvm.org.apache.iceberg.FileFormat.PARQUET) \ + .withFileSizeInBytes(table.io().newInputFile(path).getLength()) \ + .withMetrics(appender.metrics()).build() + table.newAppend().appendFile(data_file).commit() + update = table.updateSchema() + int_type = jvm.org.apache.iceberg.types.Types.IntegerType.get() + if format_version == '3': + update.addColumn(parent, 'added', int_type, + jvm.org.apache.iceberg.expressions.Literal.of(default)).commit() + else: + update.addColumn(parent, 'added', int_type).commit() + spark.sql(f'REFRESH TABLE {table_name}') + + with_cpu_session(setup) + + def read(spark): + # An unpruned CPU reference preserves parent validity for default-only projections. + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', + spark.conf.get('spark.rapids.sql.enabled') == 'true') + return spark.sql(f'SELECT id, {expression} AS added FROM {table_name}') + + def check(cpu_rows, gpu_rows): + if shape == 'map': + def normalize(rows): + return [(row[0], None if row[1] is None else sorted(row[1], key=repr)) + for row in rows] + cpu_rows, gpu_rows = normalize(cpu_rows), normalize(gpu_rows) + return _check_default_rows(expected)(cpu_rows, gpu_rows) + + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type, + 'spark.sql.iceberg.vectorization.enabled': 'false'}, + result_canonicalize_func_before_compare=check) + + +def _create_iceberg_table_from_schema(spark, table_name, fields, format_version): + jvm = spark_jvm() + schema = jvm.org.apache.iceberg.SchemaParser.fromJson(json.dumps({ + 'type': 'struct', 'fields': fields})) + catalog = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergCatalog( + spark._jsparkSession, 'spark_catalog') + props = jvm.java.util.HashMap() + for key, value in _build_tblprops({'format-version': format_version}).items(): + props.put(key, value) + return catalog.createTable(jvm.org.apache.iceberg.catalog.TableIdentifier.parse(table_name), + schema, jvm.org.apache.iceberg.PartitionSpec.unpartitioned(), props) + + +def _append_external_parquet(spark, table, path, row_count): + jvm = spark_jvm() + hpath = jvm.org.apache.hadoop.fs.Path(path) + size = hpath.getFileSystem(spark._jsc.hadoopConfiguration()).getFileStatus(hpath).getLen() + data_file = jvm.org.apache.iceberg.DataFiles.builder(table.spec()).withPath(path) \ + .withFormat(jvm.org.apache.iceberg.FileFormat.PARQUET) \ + .withFileSizeInBytes(size).withRecordCount(row_count).build() + mapping = jvm.org.apache.iceberg.mapping.MappingUtil.create(table.schema()) + table.updateProperties().set('schema.name-mapping.default', + jvm.org.apache.iceberg.mapping.NameMappingParser.toJson(mapping)).commit() + table.newAppend().appendFile(data_file).commit() + + +@iceberg +@pytest.mark.parametrize('format_version', ['2', pytest.param('3', marks=pytest.mark.skipif( + not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON))]) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@ignore_order(local=True) +def test_iceberg_unmapped_validity_helper_id( + spark_tmp_table_factory, spark_tmp_path, format_version, reader_type): + table_name = get_full_table_name(spark_tmp_table_factory) + + def setup(spark): + fields = [ + {'id': 1, 'name': 'id', 'required': False, 'type': 'int'}, + {'id': 2, 'name': 's', 'required': False, 'type': {'type': 'struct', 'fields': [ + {'id': 3, 'name': 'present', 'required': False, 'type': 'long'}]}}] + # Evolve through real fields so the new column receives ID 1000, the first ID that + # Iceberg's Parquet converter would otherwise invent for an unmapped helper leaf. + fields += [{'id': i, 'name': f'unused_{i}', 'required': False, 'type': 'int'} + for i in range(4, 1000)] + table = _create_iceberg_table_from_schema(spark, table_name, fields, format_version) + update = table.updateSchema() + for i in range(4, 1000): + update.deleteColumn(f'unused_{i}') + update.commit() + source = spark_tmp_path + '/unmapped' + spark.createDataFrame([(0, None), (1, (99, 7)), (2, (100, None))], + 'id INT, s STRUCT') \ + .coalesce(1).write.parquet(source) + jvm = spark_jvm() + path = jvm.org.apache.hadoop.fs.Path(source) + files = path.getFileSystem(spark._jsc.hadoopConfiguration()).listStatus(path) + parquet = [f.getPath().toString() for f in files + if f.getPath().getName().endswith('.parquet')] + assert len(parquet) == 1 + _append_external_parquet(spark, table, parquet[0], 3) + update = table.updateSchema() + int_type = jvm.org.apache.iceberg.types.Types.IntegerType.get() + if format_version == '3': + update.addColumn('s', 'added', int_type, + jvm.org.apache.iceberg.expressions.Literal.of(11)).commit() + else: + update.addColumn('s', 'added', int_type).commit() + assert table.schema().findField('s.added').fieldId() == 1000 + spark.sql(f'REFRESH TABLE {table_name}') + + with_cpu_session(setup, conf=disable_parquet_field_id_write) + + def read(spark): + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', + spark.conf.get('spark.rapids.sql.enabled') == 'true') + return spark.sql(f'SELECT id, s.added FROM {table_name}') + + default = 11 if format_version == '3' else None + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': str(format_version == '3').lower(), + 'spark.rapids.sql.format.parquet.reader.type': reader_type, + 'spark.sql.iceberg.vectorization.enabled': 'false'}, + result_canonicalize_func_before_compare=_check_default_rows([ + (0, None), (1, default), (2, default)])) + + +@iceberg +@pytest.mark.parametrize('format_version', ['2', pytest.param('3', marks=pytest.mark.skipif( + not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON))]) +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@ignore_order(local=True) +def test_iceberg_legacy_list_initial_defaults( + spark_tmp_table_factory, spark_tmp_path, format_version, reader_type): + table_name = get_full_table_name(spark_tmp_table_factory) + path = spark_tmp_path + '/legacy-list.parquet' + + def setup(spark): + fields = [ + {'id': 1, 'name': 'id', 'required': False, 'type': 'int'}, + {'id': 2, 'name': 'items', 'required': False, 'type': { + 'type': 'list', 'element-id': 3, 'element-required': True, + 'element': {'type': 'struct', 'fields': [ + {'id': 4, 'name': 'a', 'required': False, 'type': 'long'}, + {'id': 5, 'name': 'b', 'required': False, 'type': 'long'}]}}}] + table = _create_iceberg_table_from_schema(spark, table_name, fields, format_version) + jvm = spark_jvm() + schema = jvm.org.apache.parquet.schema.MessageTypeParser.parseMessageType( + 'message test { optional int32 id; optional group items (LIST) { ' + 'repeated group element { optional int64 a; optional int64 b; } } }') + factory = jvm.org.apache.parquet.example.data.simple.SimpleGroupFactory(schema) + writer = jvm.org.apache.parquet.hadoop.example.ExampleParquetWriter \ + .builder(jvm.org.apache.hadoop.fs.Path(path)) \ + .withConf(spark._jsc.hadoopConfiguration()).withType(schema).build() + try: + for i in range(3): + row = factory.newGroup().append('id', i) + if i > 0: + items = row.addGroup('items') + if i == 2: + items.addGroup('element') + # Values outside INT32 force Py4J to use the INT64 append overload. + items.addGroup('element').append('a', 5000000000).append('b', 6000000000) + writer.write(row) + finally: + writer.close() + _append_external_parquet(spark, table, path, 3) + update = table.updateSchema() + int_type = jvm.org.apache.iceberg.types.Types.IntegerType.get() + if format_version == '3': + update.addColumn('items.element', 'added', int_type, + jvm.org.apache.iceberg.expressions.Literal.of(11)).commit() + else: + update.addColumn('items.element', 'added', int_type).commit() + spark.sql(f'REFRESH TABLE {table_name}') + + with_cpu_session(setup) + default = 11 if format_version == '3' else None + for project_sibling in (False, True): + def read(spark): + if spark.conf.get('spark.rapids.sql.enabled') == 'true': + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', 'true') + columns = 'id, items.b AS present, items.added' if project_sibling else \ + 'id, items.added' + return spark.sql(f'SELECT {columns} FROM {table_name}') + # Iceberg's CPU row reader fails on this legacy layout. Spark's Parquet reader + # provides an independent reference, including null and empty list cardinality. + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', 'false') + literal = '11' if format_version == '3' else 'NULL' + columns = ['id'] + (['items.b AS present'] if project_sibling else []) + columns += [f'transform(items, x -> CAST({literal} AS INT)) AS added'] + return spark.read.parquet(path).selectExpr(*columns) + + expected = ([(0, None, None), (1, [], []), (2, [None, 6000000000], [default, default])] + if project_sibling else [(0, None), (1, []), (2, [default, default])]) + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': str(format_version == '3').lower(), + 'spark.rapids.sql.format.parquet.reader.type': reader_type}, + result_canonicalize_func_before_compare=_check_default_rows(expected)) + + +@iceberg +@pytest.mark.skipif( + not is_spark_400_or_later() or not supports_iceberg_v3 or + tuple(int(part) for part in runtime_iceberg_version.split('.')[:2]) < (1, 11), + reason='Iceberg VARIANT writes require Spark 4.x and Iceberg 1.11 or later') +@pytest.mark.parametrize('reader_type', rapids_reader_types) +@ignore_order(local=True) +def test_iceberg_v3_default_with_unprojected_variant(spark_tmp_table_factory, reader_type): + table_name = get_full_table_name(spark_tmp_table_factory) + + def setup(spark): + props = _build_tblprops({'format-version': '3'}) + props_sql = ', '.join(f"'{key}' = '{value}'" for key, value in props.items()) + spark.sql(f'CREATE TABLE {table_name} (id INT, s STRUCT) ' + f'USING ICEBERG TBLPROPERTIES ({props_sql})') + # Distinguish a null parent, SQL-null Variant, JSON null, and a non-null object. + spark.sql( + 'SELECT id, CASE WHEN id = 0 THEN NULL ELSE ' + "named_struct('present', parse_json(text)) END AS s " + "FROM VALUES (0, 'null'), (1, CAST(NULL AS STRING)), " + "(2, 'null'), (3, '{\"a\":1}') AS v(id, text)") \ + .coalesce(1).writeTo(table_name).append() + jvm = spark_jvm() + table = jvm.org.apache.iceberg.spark.Spark3Util.loadIcebergTable( + spark._jsparkSession, table_name) + table.updateSchema().addColumn( + 's', 'added', jvm.org.apache.iceberg.types.Types.IntegerType.get(), + jvm.org.apache.iceberg.expressions.Literal.of(11)).commit() + spark.sql(f'REFRESH TABLE {table_name}') + + with_cpu_session(setup) + + def read(spark): + spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', + spark.conf.get('spark.rapids.sql.enabled') == 'true') + return spark.sql(f'SELECT id, s.added FROM {table_name}') + + assert_gpu_and_cpu_are_equal_collect( + read, + conf={'spark.rapids.sql.format.iceberg.v3.enabled': 'true', + 'spark.rapids.sql.format.parquet.reader.type': reader_type, + 'spark.sql.iceberg.vectorization.enabled': 'false'}, + result_canonicalize_func_before_compare=_check_default_rows([ + (0, None), (1, 11), (2, 11), (3, 11)])) diff --git a/tests/src/test/spark350/scala/com/nvidia/spark/rapids/iceberg/GpuPostProcessorSuite.scala b/tests/src/test/spark350/scala/com/nvidia/spark/rapids/iceberg/GpuPostProcessorSuite.scala index 26c13526d19..fa0c82f2022 100644 --- a/tests/src/test/spark350/scala/com/nvidia/spark/rapids/iceberg/GpuPostProcessorSuite.scala +++ b/tests/src/test/spark350/scala/com/nvidia/spark/rapids/iceberg/GpuPostProcessorSuite.scala @@ -28,7 +28,8 @@ spark-rapids-shim-json-lines ***/ package com.nvidia.spark.rapids.iceberg -import java.util.{HashMap => JHashMap} +import java.nio.ByteBuffer +import java.util.{HashMap => JHashMap, UUID} import scala.collection.JavaConverters._ @@ -40,11 +41,15 @@ import com.nvidia.spark.rapids.parquet.ParquetFileInfoWithBlockMeta import com.nvidia.spark.rapids.spill.SpillFramework import org.apache.hadoop.fs.Path import org.apache.iceberg.{MetadataColumns, Schema} +import org.apache.iceberg.expressions.Expressions +import org.apache.iceberg.parquet.ParquetSchemaUtil import org.apache.iceberg.shaded.org.apache.parquet.schema.{ MessageType => ShadedMessageType, + MessageTypeParser => ShadedMessageTypeParser, Type => ShadedType, Types => ShadedTypes } +import org.apache.iceberg.shaded.org.apache.parquet.schema.LogicalTypeAnnotation import org.apache.iceberg.shaded.org.apache.parquet.schema.PrimitiveType.{ PrimitiveTypeName => ShadedPrimitiveTypeName } @@ -56,7 +61,8 @@ import org.scalatest.funsuite.AnyFunSuite import org.apache.spark.SparkConf import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.types.{LongType, StructType} +import org.apache.spark.sql.execution.datasources.parquet.ParquetToSparkSchemaConverter +import org.apache.spark.sql.types.{BinaryType, DateType, LongType, StructType} import org.apache.spark.sql.vectorized.ColumnarBatch /** @@ -65,6 +71,30 @@ import org.apache.spark.sql.vectorized.ColumnarBatch */ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll { + private def fieldWithDefaults( + id: Int, + name: String, + icebergType: org.apache.iceberg.types.Type, + required: Boolean, + initialDefault: Option[AnyRef]): Option[Types.NestedField] = { + try { + val factory = classOf[Types.NestedField].getMethod( + if (required) "required" else "optional", classOf[String]) + val builder = factory.invoke(null, name) + val builderClass = builder.getClass + builderClass.getMethod("withId", java.lang.Integer.TYPE) + .invoke(builder, Int.box(id)) + builderClass.getMethod("ofType", classOf[org.apache.iceberg.types.Type]) + .invoke(builder, icebergType) + initialDefault.foreach { value => + builderClass.getMethod("withInitialDefault", classOf[Object]).invoke(builder, value) + } + Some(builderClass.getMethod("build").invoke(builder).asInstanceOf[Types.NestedField]) + } catch { + case _: NoSuchMethodException => None + } + } + private lazy val supportsRowLineageInheritance = { val majorMinor = IcebergProvider.detectedVersion.split("\\.").take(2).mkString(".") Set("1.10", "1.11").contains(majorMinor) @@ -909,6 +939,585 @@ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll { } } + test("Missing required primitive is filled with its Iceberg initial default") { + import com.nvidia.spark.rapids.Arm.withResource + import com.nvidia.spark.rapids.GpuColumnVector + import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector => SparkColumnVector} + + val fieldId = 1 + val defaultField = fieldWithDefaults( + fieldId, + "added", + Types.IntegerType.get(), + required = true, + initialDefault = Some(Int.box(7))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val parquetSchema = new ShadedMessageType("test", Seq.empty[ShadedType].asJava) + val (parquetInfo, shadedSchema) = createParquetInfo(parquetSchema) + val processor = new GpuParquetReaderPostProcessor( + parquetInfo, + new JHashMap[Integer, Any](), + new Schema(defaultField), + shadedSchema, + Map.empty) + + assert(processor.displayActionPlan().contains( + s"FillDefault(fieldId=$fieldId, int)")) + + withResource(processor.process(new ColumnarBatch(Array.empty[SparkColumnVector], 3))) { + outputBatch => + withResource(outputBatch.column(0).asInstanceOf[GpuColumnVector].copyToHost()) { hostCol => + (0 until outputBatch.numRows()).foreach { row => + assert(hostCol.getBase.getInt(row) == 7) + } + } + } + } + + test("Unsupported Iceberg defaults are identified by subtype and nested path") { + val timestampNtzField = fieldWithDefaults( + 3, + "created_at", + Types.TimestampType.withoutZone(), + required = false, + initialDefault = Some(Long.box(0L))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val timestampField = fieldWithDefaults( + 4, + "updated_at", + Types.TimestampType.withZone(), + required = false, + initialDefault = Some(Long.box(0L))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val uuidField = fieldWithDefaults( + 5, + "uuid_value", + Types.UUIDType.get(), + required = false, + initialDefault = Some(UUID.fromString("123e4567-e89b-12d3-a456-426614174000"))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val fixedType = Types.FixedType.ofLength(2) + val fixedField = fieldWithDefaults( + 6, + "fixed_value", + fixedType, + required = false, + initialDefault = Some(ByteBuffer.wrap(Array[Byte](1, 2)))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + + val unsupportedSchema = new Schema(Types.NestedField.optional( + 1, "payload", Types.StructType.of(timestampNtzField))) + val unsupportedListSchema = new Schema(Types.NestedField.optional( + 1, "events", Types.ListType.ofOptional( + 2, Types.StructType.of(timestampNtzField)))) + val unsupportedMapKeySchema = new Schema(Types.NestedField.optional( + 1, "lookup", Types.MapType.ofOptional( + 2, 4, Types.StructType.of(timestampNtzField), Types.StringType.get()))) + val unsupportedMapValueSchema = new Schema(Types.NestedField.optional( + 1, "lookup", Types.MapType.ofOptional( + 2, 4, Types.StringType.get(), Types.StructType.of(timestampNtzField)))) + val supportedSchema = new Schema(timestampField) + + assert(IcebergFormatVersionSupport.unsupportedDefault(unsupportedSchema) + .contains("payload.created_at" -> "timestamp")) + assert(IcebergFormatVersionSupport.unsupportedDefault(unsupportedListSchema) + .contains("events.element.created_at" -> "timestamp")) + assert(IcebergFormatVersionSupport.unsupportedDefault(unsupportedMapKeySchema) + .contains("lookup.key.created_at" -> "timestamp")) + assert(IcebergFormatVersionSupport.unsupportedDefault(unsupportedMapValueSchema) + .contains("lookup.value.created_at" -> "timestamp")) + assert(IcebergFormatVersionSupport.unsupportedDefault(supportedSchema).isEmpty) + assert(IcebergFormatVersionSupport.unsupportedDefault(new Schema(uuidField)) + .contains("uuid_value" -> Types.UUIDType.get().toString)) + assert(IcebergFormatVersionSupport.unsupportedDefault(new Schema(fixedField)) + .contains("fixed_value" -> fixedType.toString)) + } + + test("Equality-delete required fields are included in the unsupported default check") { + val projectedField = Types.NestedField.optional(1, "id", Types.LongType.get()) + val equalityDeleteField = fieldWithDefaults( + 2, + "delete_key", + Types.TimestampType.withoutZone(), + required = false, + initialDefault = Some(Long.box(0L))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val unreferencedField = fieldWithDefaults( + 3, + "unreferenced", + Types.TimestampType.withoutZone(), + required = false, + initialDefault = Some(Long.box(0L))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val tableSchema = new Schema(projectedField, equalityDeleteField, unreferencedField) + val expectedSchema = new Schema(projectedField) + + val schemaToCheck = IcebergFormatVersionSupport.withRequiredFields( + tableSchema, expectedSchema, Seq(equalityDeleteField.fieldId())) + + assert(schemaToCheck.columns().asScala.map(_.fieldId()) == Seq(1, 2)) + assert(IcebergFormatVersionSupport.unsupportedDefault(schemaToCheck) + .contains("delete_key" -> "timestamp")) + } + + test("Nested equality-delete required fields fall back whether projected or implicit") { + val projectedField = Types.NestedField.optional(1, "id", Types.LongType.get()) + val equalityDeleteField = fieldWithDefaults( + 3, + "delete_key", + Types.StringType.get(), + required = false, + initialDefault = Some("legacy")).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val unreferencedField = fieldWithDefaults( + 4, + "unreferenced", + Types.TimestampType.withoutZone(), + required = false, + initialDefault = Some(Long.box(0L))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val payloadField = Types.NestedField.optional( + 2, "payload", Types.StructType.of(equalityDeleteField)) + val tableSchema = new Schema(projectedField, payloadField, unreferencedField) + + Seq( + new Schema(projectedField), + new Schema(projectedField, payloadField)).foreach { expectedSchema => + val error = intercept[UnsupportedOperationException] { + IcebergFormatVersionSupport.withRequiredFields( + tableSchema, expectedSchema, Seq(equalityDeleteField.fieldId())) + } + + assert(error.getMessage.contains( + "Nested equality-delete field 'delete_key' with ID 3 is not supported on GPU")) + } + } + + test("Present struct children take precedence over defaults for missing children") { + val structId = 1 + val presentId = 2 + val defaultId = 3 + val defaultField = fieldWithDefaults( + defaultId, + "added", + Types.IntegerType.get(), + required = true, + initialDefault = Some(Int.box(11))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + + val presentParquetField = + ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.OPTIONAL) + .id(presentId).named("present") + val parquetStruct = ShadedTypes.optionalGroup().addField(presentParquetField) + .id(structId).named("s") + val parquetSchema = new ShadedMessageType("test", Seq[ShadedType](parquetStruct).asJava) + val expectedSchema = new Schema( + Types.NestedField.optional(structId, "s", Types.StructType.of( + Types.NestedField.optional(presentId, "present", Types.LongType.get()), + defaultField))) + val (parquetInfo, shadedSchema) = createParquetInfo(parquetSchema) + val processor = new GpuParquetReaderPostProcessor( + parquetInfo, + new JHashMap[Integer, Any](), + expectedSchema, + shadedSchema, + Map.empty) + + val plan = processor.displayActionPlan() + assert(plan.contains("field[0] (input[0]):\n PassThrough"), plan) + assert(plan.contains(s"FillDefault(fieldId=$defaultId, int)"), plan) + } + + test("Row-group pruning preserves missing defaults and recognizes physical field IDs") { + val defaultField = fieldWithDefaults( + 2, "added", Types.IntegerType.get(), required = false, + initialDefault = Some(Int.box(11))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val nestedDefault = fieldWithDefaults( + 4, "added", Types.IntegerType.get(), required = false, + initialDefault = Some(Int.box(11))).get + val requiredSchema = new Schema( + Types.NestedField.optional(1, "id", Types.LongType.get()), + defaultField, + Types.NestedField.optional(3, "s", Types.StructType.of(nestedDefault))) + val physicalId = ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).id(1).named("id") + val physicalSibling = + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).id(5).named("present") + val physicalStruct = ShadedTypes.optionalGroup().addField(physicalSibling).id(3).named("s") + val oldSchema = new ShadedMessageType( + "test", Seq[ShadedType](physicalId, physicalStruct).asJava) + + Seq( + Expressions.equal("added", 11), + Expressions.greaterThan("added", 10), + Expressions.notNull("added"), + Expressions.not(Expressions.equal("added", 23)), + Expressions.and(Expressions.equal("id", 1L), Expressions.equal("added", 11)), + Expressions.or(Expressions.equal("id", 1L), Expressions.equal("added", 11)), + Expressions.equal("s.added", 11)).foreach { filter => + assert(GpuIcebergParquetReader.hasMissingDefaultFilter( + oldSchema, requiredSchema, filter, caseSensitive = true), filter.toString) + } + assert(!GpuIcebergParquetReader.hasMissingDefaultFilter( + oldSchema, requiredSchema, Expressions.equal("id", 1L), caseSensitive = true)) + assert(GpuIcebergParquetReader.hasMissingDefaultFilter( + oldSchema, requiredSchema, Expressions.equal("ADDED", 11), caseSensitive = false)) + + val renamedPhysical = + ShadedTypes.optional(ShadedPrimitiveTypeName.INT32).id(2).named("old_name") + val currentSchema = new ShadedMessageType( + "test", Seq[ShadedType](physicalId, renamedPhysical, physicalStruct).asJava) + assert(!GpuIcebergParquetReader.hasMissingDefaultFilter( + currentSchema, requiredSchema, Expressions.notNull("added"), caseSensitive = true)) + val nullDefaultSchema = new Schema( + Types.NestedField.optional(2, "added", Types.IntegerType.get())) + assert(!GpuIcebergParquetReader.hasMissingDefaultFilter( + oldSchema, nullDefaultSchema, Expressions.notNull("added"), caseSensitive = true)) + } + + test("Validity helpers decode physical types and preserve projected logical types") { + val dateField = ShadedTypes.optional(ShadedPrimitiveTypeName.INT32) + .as(LogicalTypeAnnotation.dateType()).id(4).named("date") + val expectedSchema = new Schema( + Types.NestedField.optional(1, "s", Types.StructType.of( + Types.NestedField.optional(3, "added", Types.IntegerType.get()))), + Types.NestedField.optional(4, "date", Types.DateType.get())) + val helpers = Seq( + ShadedTypes.optional(ShadedPrimitiveTypeName.FIXED_LEN_BYTE_ARRAY).length(16) + .as(LogicalTypeAnnotation.uuidType()).id(2).named("legacy") -> BinaryType, + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64) + .as(LogicalTypeAnnotation.timeType(false, LogicalTypeAnnotation.TimeUnit.MICROS)) + .id(2).named("legacy") -> LongType, + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64) + .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.NANOS)) + .id(2).named("legacy") -> LongType, + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64) + .as(LogicalTypeAnnotation.timestampType(false, LogicalTypeAnnotation.TimeUnit.MICROS)) + .id(2).named("legacy") -> LongType) + + helpers.foreach { case (helper, sparkType) => + val fileSchema = new ShadedMessageType("test", Seq[ShadedType]( + ShadedTypes.optionalGroup().addField(helper).id(1).named("s"), dateField).asJava) + val projected = ParquetSchemaUtil.pruneColumns(fileSchema, expectedSchema) + val retained = GpuIcebergParquetReader.preserveStructValidity( + fileSchema, projected, expectedSchema) + val physical = GpuIcebergParquetReader.physicalHelperSchema(retained, expectedSchema) + val converted = new ParquetToSparkSchemaConverter(false, true, true, true, false) + .convert(ShimUtils.getParquetUnshader().unshade(physical)) + assert(converted("s").dataType.asInstanceOf[StructType]("legacy").dataType == sparkType) + assert(converted("date").dataType == DateType) + assert(physical.getType("s", "legacy").getId.intValue() == 2) + assert(physical.asGroupType().getType("date") == dateField) + } + } + + test("Opaque logical groups used as validity helpers decode physically") { + val variant = try { + classOf[LogicalTypeAnnotation].getMethod("variantType", java.lang.Byte.TYPE) + .invoke(null, Byte.box(1.toByte)).asInstanceOf[LogicalTypeAnnotation] + } catch { + case _: NoSuchMethodException => cancel("Parquet runtime does not expose VARIANT") + } + val helper = ShadedTypes.optionalGroup() + .addField(ShadedTypes.required(ShadedPrimitiveTypeName.BINARY).named("metadata")) + .addField(ShadedTypes.required(ShadedPrimitiveTypeName.BINARY).named("value")) + .as(variant).id(2).named("legacy") + val fileSchema = new ShadedMessageType("test", Seq[ShadedType]( + ShadedTypes.optionalGroup().addField(helper).id(1).named("s")).asJava) + val defaultField = fieldWithDefaults( + 3, "added", Types.IntegerType.get(), required = false, + initialDefault = Some(Int.box(11))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val expectedSchema = new Schema( + Types.NestedField.optional(1, "s", Types.StructType.of(defaultField))) + val retained = GpuIcebergParquetReader.preserveStructValidity( + fileSchema, ParquetSchemaUtil.pruneColumns(fileSchema, expectedSchema), expectedSchema) + val physical = GpuIcebergParquetReader.physicalHelperSchema(retained, expectedSchema) + val physicalHelper = physical.getType("s", "legacy").asGroupType() + assert(physicalHelper.getLogicalTypeAnnotation == null) + assert(physicalHelper.getId.intValue() == 2) + assert(physicalHelper.getFieldCount == 1) + assert(helper.getLogicalTypeAnnotation == variant) + val converted = new ParquetToSparkSchemaConverter(false, true, true, true, false) + .convert(ShimUtils.getParquetUnshader().unshade(physical)) + assert(converted("s").dataType.asInstanceOf[StructType]("legacy") + .dataType.asInstanceOf[StructType]("metadata").dataType == BinaryType) + val (info, shaded) = createParquetInfo(physical) + val processor = new GpuParquetReaderPostProcessor( + info, new JHashMap[Integer, Any](), expectedSchema, shaded, Map.empty) + assert(processor.displayActionPlan().contains("FillDefault(fieldId=3, int)")) + + val projectedVariant = org.apache.iceberg.SchemaParser.fromJson( + """{"type":"struct","fields":[{"id":1,"name":"s","required":false, + |"type":{"type":"struct","fields":[{"id":2,"name":"legacy", + |"required":false,"type":"variant"}]}}]}""".stripMargin) + val projected = GpuIcebergParquetReader.physicalHelperSchema(fileSchema, projectedVariant) + assert(projected.getType("s", "legacy").getLogicalTypeAnnotation == variant) + } + + test("Unmapped validity helpers cannot bind to requested field IDs") { + val defaultField = fieldWithDefaults( + 1000, "added", Types.IntegerType.get(), required = false, + initialDefault = Some(Int.box(11))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val expectedSchema = new Schema( + Types.NestedField.optional(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "s", Types.StructType.of(defaultField)), + Types.NestedField.optional(3, "other", Types.LongType.get())) + val helpers = Seq( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).named("untracked"), + ShadedTypes.optionalGroup().addField( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).named("leaf")).named("untracked"), + ShadedTypes.optionalList().element(ShadedTypes.optionalGroup().addField( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).named("leaf")).named("element")) + .named("untracked"), + ShadedTypes.optionalMap().key( + ShadedTypes.required(ShadedPrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()).named("key")) + .value(ShadedTypes.optionalGroup().addField( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).named("leaf")).named("value")) + .named("untracked")) + helpers.foreach { helper => + val fileSchema = new ShadedMessageType("test", Seq[ShadedType]( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT32).id(1).named("id"), + ShadedTypes.optionalGroup().addField(helper).id(2).named("s")).asJava) + val retained = ShimUtils.pruneColumns(fileSchema, expectedSchema, false) + val physical = GpuIcebergParquetReader.physicalHelperSchema(retained, expectedSchema) + val partner = ParquetSchemaUtil.convert(physical) + assert(partner.findField(1000) == null) + assert(partner.findField(3) == null) + assert(partner.findField("s.untracked").fieldId() != 1000) + if (helper.getLogicalTypeAnnotation != null) { + assert(physical.getType("s", "untracked").asGroupType().getType(0).getId == null) + } + val (info, shaded) = createParquetInfo(physical) + val processor = new GpuParquetReaderPostProcessor( + info, new JHashMap[Integer, Any](), expectedSchema, shaded, Map.empty) + assert(processor.displayActionPlan().contains("FillDefault(fieldId=1000, int)")) + } + } + + test("Legacy two-level LIST projections preserve struct element interpretation") { + val defaultField = fieldWithDefaults( + 5, "added", Types.IntegerType.get(), required = false, + initialDefault = Some(Int.box(11))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + for (elementName <- Seq("element", "array"); projectSibling <- Seq(false, true)) { + val fileSchema = ShadedMessageTypeParser.parseMessageType( + s"""message test { + | optional group items (LIST) = 1 { + | repeated group $elementName = 2 { + | optional int64 a = 3; + | optional int64 b = 4; + | } + | } + |}""".stripMargin) + val fields = if (projectSibling) { + Seq(Types.NestedField.optional(3, "a", Types.LongType.get()), defaultField) + } else { + Seq(defaultField) + } + val expectedSchema = new Schema(Types.NestedField.optional(1, "items", + Types.ListType.ofRequired(2, Types.StructType.of(fields.asJava)))) + val retained = GpuIcebergParquetReader.preserveStructValidity( + fileSchema, ParquetSchemaUtil.pruneColumns(fileSchema, expectedSchema), expectedSchema) + val physical = GpuIcebergParquetReader.physicalHelperSchema(retained, expectedSchema) + val element = ParquetSchemaUtil.convert(physical).findType("items") + .asListType().elementType() + assert(element.isStructType) + assert(element.asStructType().fields().size() == (if (elementName == "element") 2 else 1)) + val (info, shaded) = createParquetInfo(physical) + val processor = new GpuParquetReaderPostProcessor( + info, new JHashMap[Integer, Any](), expectedSchema, shaded, Map.empty) + assert(processor.displayActionPlan().contains("FillDefault(fieldId=5, int)")) + } + } + + test("Default-only struct projections retain physical validity through nested containers") { + val defaultField = fieldWithDefaults( + 3, + "added", + Types.IntegerType.get(), + required = false, + initialDefault = Some(Int.box(11))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val expectedStruct = Types.StructType.of(defaultField) + val presentField = + ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.OPTIONAL) + .id(2).named("present") + val ignoredField = + ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.OPTIONAL) + .id(6).named("ignored") + val keyField = + ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.REQUIRED) + .id(5).named("key") + val unrelatedField = + ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.OPTIONAL) + .id(9).named("unrelated") + + def physicalStruct(name: String, includeIgnored: Boolean) = { + val builder = ShadedTypes.optionalGroup().addField(presentField) + if (includeIgnored) { + builder.addField(ignoredField) + } + builder.id(1).named(name) + } + + def physicalFields(includeIgnored: Boolean): Seq[ShadedType] = Seq( + physicalStruct("s", includeIgnored), + ShadedTypes.optionalGroup().addField(physicalStruct("s", includeIgnored)) + .id(4).named("outer"), + ShadedTypes.optionalList().element(physicalStruct("element", includeIgnored)) + .id(4).named("items"), + ShadedTypes.optionalMap().key(keyField).value(physicalStruct("value", includeIgnored)) + .id(4).named("entries"), + ShadedTypes.optionalGroup().addField( + ShadedTypes.optionalMap().key(keyField).value(physicalStruct("value", includeIgnored)) + .id(7).named("entries")) + .id(4).named("outer"), + ShadedTypes.optionalGroup().addField( + ShadedTypes.optionalList().element(physicalStruct("element", includeIgnored)) + .id(7).named("items")) + .id(4).named("outer")) + + val expectedFields = Seq( + Types.NestedField.optional(1, "s", expectedStruct), + Types.NestedField.optional(4, "outer", Types.StructType.of( + Types.NestedField.optional(1, "s", expectedStruct))), + Types.NestedField.optional(4, "items", Types.ListType.ofOptional(1, expectedStruct)), + Types.NestedField.optional(4, "entries", + Types.MapType.ofOptional(5, 1, Types.LongType.get(), expectedStruct)), + Types.NestedField.optional(4, "outer", Types.StructType.of( + Types.NestedField.optional(7, "entries", + Types.MapType.ofOptional(5, 1, Types.LongType.get(), expectedStruct)))), + Types.NestedField.optional(4, "outer", Types.StructType.of( + Types.NestedField.optional(7, "items", Types.ListType.ofOptional(1, expectedStruct))))) + + physicalFields(includeIgnored = true).zip(physicalFields(includeIgnored = false)) + .zip(expectedFields).foreach { case ((fileField, retainedField), expectedField) => + withClue(s"${expectedField.name()}: ") { + val fileSchema = new ShadedMessageType("test", Seq(fileField, unrelatedField).asJava) + val expectedSchema = new Schema(expectedField) + val projected = ParquetSchemaUtil.pruneColumns(fileSchema, expectedSchema) + val restored = GpuIcebergParquetReader.preserveStructValidity( + fileSchema, projected, expectedSchema) + val projectedPaths = projected.getPaths.asScala.map(_.toSeq).toSet + val helperSchema = new ShadedMessageType("test", Seq(retainedField).asJava) + val helperPaths = helperSchema.getPaths.asScala.map(_.toSeq).toSet + // Iceberg can already retain extra physical siblings for nested projections. + // Preserve those fields and add only the helper paths needed by empty structs. + assert(restored.getPaths.asScala.map(_.toSeq).toSet == projectedPaths ++ helperPaths) + val partnerSchema = ParquetSchemaUtil.convert(restored) + assert(partnerSchema.findField(1) != null) + assert(partnerSchema.findField(2) != null) + assert((partnerSchema.findField(6) != null) == + projectedPaths.exists(_.last == "ignored")) + assert(partnerSchema.findField(9) == null) + assert(GpuIcebergParquetReader.preserveStructValidity( + fileSchema, fileSchema, expectedSchema) == fileSchema) + val emptySchema = new ShadedMessageType("test", Seq.empty[ShadedType].asJava) + assert(GpuIcebergParquetReader.preserveStructValidity( + fileSchema, emptySchema, new Schema()) == emptySchema) + + val (parquetInfo, shadedSchema) = createParquetInfo(restored) + val processor = new GpuParquetReaderPostProcessor( + parquetInfo, + new JHashMap[Integer, Any](), + expectedSchema, + shadedSchema, + Map.empty) + val plan = processor.displayActionPlan() + assert(plan.contains("FillDefault(fieldId=3, int)"), plan) + assert(!plan.contains("FillNull"), plan) + } + } + } + + test("Default-only projections restore all requested sibling collection paths") { + def addedField(id: Int) = fieldWithDefaults( + id, "added", Types.IntegerType.get(), required = false, + initialDefault = Some(Int.box(11))).getOrElse { + cancel("Iceberg runtime does not expose v3 field defaults") + } + val unrelated = ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).id(2).named("unrelated") + val listElement = ShadedTypes.optionalGroup().addField( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).id(5).named("present")) + .id(4).named("element") + val items = ShadedTypes.optionalList().element(listElement).id(3).named("b") + val mapKey = ShadedTypes.required(ShadedPrimitiveTypeName.INT64).id(8).named("key") + val mapValue = ShadedTypes.optionalGroup().addField( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).id(10).named("present")) + .id(9).named("value") + val entries = ShadedTypes.optionalMap().key(mapKey).value(mapValue).id(7).named("c") + val fileStruct = ShadedTypes.optionalGroup().addFields(unrelated, items, entries) + .id(1).named("s") + val requestedCollections = Seq( + Types.NestedField.optional(3, "b", + Types.ListType.ofOptional(4, Types.StructType.of(addedField(6)))), + Types.NestedField.optional(7, "c", Types.MapType.ofOptional( + 8, 9, Types.LongType.get(), Types.StructType.of(addedField(11))))) + + for (nested <- Seq(false, true); projectSibling <- Seq(false, true)) { + withClue(s"nested=$nested, projectSibling=$projectSibling: ") { + val fields = if (projectSibling) { + Types.NestedField.optional(2, "unrelated", Types.LongType.get()) +: + requestedCollections + } else { + requestedCollections + } + val requiredStruct = Types.NestedField.optional(1, "s", Types.StructType.of(fields.asJava)) + val fileField = if (nested) { + ShadedTypes.optionalGroup().addField( + ShadedTypes.optional(ShadedPrimitiveTypeName.INT64).id(13).named("first")) + .addField(fileStruct).id(12).named("outer") + } else { + fileStruct + } + val requiredField = if (nested) { + Types.NestedField.optional(12, "outer", Types.StructType.of(requiredStruct)) + } else { + requiredStruct + } + val fileSchema = new ShadedMessageType("test", Seq[ShadedType](fileField).asJava) + val requiredSchema = new Schema(requiredField) + val projected = ParquetSchemaUtil.pruneColumns(fileSchema, requiredSchema) + val restored = GpuIcebergParquetReader.preserveStructValidity( + fileSchema, projected, requiredSchema) + val partner = ParquetSchemaUtil.convert(restored) + val projectedPaths = projected.getPaths.asScala.map(_.toSeq).toSet + Seq(1, 3, 4, 5, 7, 8, 9, 10).foreach(id => assert(partner.findField(id) != null)) + // Keep extra siblings already retained by Iceberg, but do not add unrelated helpers. + assert((partner.findField(2) != null) == projectedPaths.exists(_.last == "unrelated")) + assert((partner.findField(13) != null) == projectedPaths.exists(_.last == "first")) + assert(GpuIcebergParquetReader.preserveStructValidity( + fileSchema, restored, requiredSchema) == restored) + + val (parquetInfo, shadedSchema) = createParquetInfo(restored) + val processor = new GpuParquetReaderPostProcessor( + parquetInfo, new JHashMap[Integer, Any](), requiredSchema, shadedSchema, Map.empty) + val plan = processor.displayActionPlan() + assert(plan.contains("FillDefault(fieldId=6, int)"), plan) + assert(plan.contains("FillDefault(fieldId=11, int)"), plan) + assert(!plan.contains("FillNull"), plan) + } + } + } + test("Whole-struct constants build a FetchConstant action") { val structFieldId = 1 val fieldAId = 2