diff --git a/dist/unshimmed-common-from-single-shim.txt b/dist/unshimmed-common-from-single-shim.txt index 15a16b0f640..f11f3531d14 100644 --- a/dist/unshimmed-common-from-single-shim.txt +++ b/dist/unshimmed-common-from-single-shim.txt @@ -9,31 +9,11 @@ com/nvidia/spark/rapids/optimizer/SQLOptimizerPlugin* com/nvidia/spark/rapids/ShimLoaderTemp* com/nvidia/spark/rapids/SparkShims* com/nvidia/shaded/spark/com/google/flatbuffers/*.class -com/nvidia/spark/rapids/fileio/iceberg/IcebergInputFile.class -com/nvidia/spark/rapids/fileio/iceberg/IcebergInputStream.class -com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputFile.class -com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputStream.class -com/nvidia/spark/rapids/iceberg/GpuInternalRowBase.class -com/nvidia/spark/rapids/iceberg/data/GpuDeleteFilter2.class -com/nvidia/spark/rapids/iceberg/package.class -com/nvidia/spark/rapids/iceberg/package$.class -com/nvidia/spark/rapids/iceberg/parquet/FileSchemaAccessors.class -com/nvidia/spark/rapids/iceberg/parquet/GpuIcebergParquetReader$.class -com/nvidia/spark/rapids/iceberg/parquet/SingleFile.class -com/nvidia/spark/rapids/iceberg/parquet/SingleFile$.class -com/nvidia/spark/rapids/iceberg/parquet/ThreadConf.class com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.class com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.class com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.class -org/apache/iceberg/aws/s3/IcebergS3InputFileAccess.class -org/apache/iceberg/data/GpuFileHelpers.class -org/apache/iceberg/io/GpuClusteredWriterBridge.class -org/apache/iceberg/io/GpuFanoutWriterBridge.class -org/apache/iceberg/io/GpuPositionDeleteFileWriter$.class -org/apache/iceberg/parquet/GpuParquetIOAccess.class org/apache/iceberg/spark/source/GpuBaseReader.class org/apache/iceberg/spark/source/GpuSparkPlanningUtil.class -org/apache/iceberg/spark/source/GpuSparkScanAccess.class org/apache/iceberg/spark/source/GpuSparkWriteAccess.class org/apache/spark/sql/rapids/AdaptiveSparkPlanHelperShim* org/apache/spark/sql/rapids/ExecutionPlanCaptureCallback* diff --git a/iceberg-common/pom.xml b/iceberg-common/pom.xml index 2ec7f766312..57520e19d74 100644 --- a/iceberg-common/pom.xml +++ b/iceberg-common/pom.xml @@ -40,6 +40,11 @@ + + com.nvidia + spark-rapids-jni + ${jni.classifier} + org.apache.iceberg iceberg-spark-runtime-${iceberg.artifact.suffix}_${scala.binary.version} diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputFile.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputFile.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputFile.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputFile.java diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputStream.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputStream.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputStream.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergInputStream.java diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputFile.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputFile.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputFile.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputFile.java diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputStream.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputStream.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputStream.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergOutputStream.java diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/GpuInternalRowBase.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/GpuInternalRowBase.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/GpuInternalRowBase.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/GpuInternalRowBase.java diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergDeletionVector.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergDeletionVector.java similarity index 98% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergDeletionVector.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergDeletionVector.java index d04b03310ff..859805ab427 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergDeletionVector.java +++ b/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergDeletionVector.java @@ -27,7 +27,8 @@ import java.util.zip.CRC32; /** - * An Iceberg deletion vector kept in its compressed Roaring-bitmap representation. + * An Iceberg deletion vector kept in its compressed Roaring-bitmap representation. This helper + * does not depend on Iceberg runtime classes. * *

The serialized bytes use the portable 64-bit Roaring format expected by cuDF. This object * owns its host buffer and must be closed after all borrowed references have been released. diff --git a/iceberg/common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFileAccess.java b/iceberg-common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFileAccess.java similarity index 100% rename from iceberg/common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFileAccess.java rename to iceberg-common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFileAccess.java 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 similarity index 100% rename from iceberg/common/src/main/java/org/apache/iceberg/data/GpuFileHelpers.java rename to iceberg-common/src/main/java/org/apache/iceberg/data/GpuFileHelpers.java diff --git a/iceberg/common/src/main/java/org/apache/iceberg/io/GpuClusteredWriterBridge.java b/iceberg-common/src/main/java/org/apache/iceberg/io/GpuClusteredWriterBridge.java similarity index 100% rename from iceberg/common/src/main/java/org/apache/iceberg/io/GpuClusteredWriterBridge.java rename to iceberg-common/src/main/java/org/apache/iceberg/io/GpuClusteredWriterBridge.java diff --git a/iceberg/common/src/main/java/org/apache/iceberg/io/GpuFanoutWriterBridge.java b/iceberg-common/src/main/java/org/apache/iceberg/io/GpuFanoutWriterBridge.java similarity index 100% rename from iceberg/common/src/main/java/org/apache/iceberg/io/GpuFanoutWriterBridge.java rename to iceberg-common/src/main/java/org/apache/iceberg/io/GpuFanoutWriterBridge.java diff --git a/iceberg/common/src/main/java/org/apache/iceberg/parquet/GpuParquetIOAccess.java b/iceberg-common/src/main/java/org/apache/iceberg/parquet/GpuParquetIOAccess.java similarity index 100% rename from iceberg/common/src/main/java/org/apache/iceberg/parquet/GpuParquetIOAccess.java rename to iceberg-common/src/main/java/org/apache/iceberg/parquet/GpuParquetIOAccess.java diff --git a/iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkScanAccess.java b/iceberg-common/src/main/java/org/apache/iceberg/spark/source/GpuSparkScanAccess.java similarity index 100% rename from iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkScanAccess.java rename to iceberg-common/src/main/java/org/apache/iceberg/spark/source/GpuSparkScanAccess.java diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/package.scala b/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/package.scala similarity index 100% rename from iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/package.scala rename to iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/package.scala diff --git a/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/FileSchemaAccessors.scala b/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/FileSchemaAccessors.scala new file mode 100644 index 00000000000..f24514444ef --- /dev/null +++ b/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/FileSchemaAccessors.scala @@ -0,0 +1,49 @@ +/* + * Copyright (c) 2025-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 org.apache.iceberg.schema.SchemaWithPartnerVisitor +import org.apache.iceberg.types.Type + +/** + * Partner accessors to navigate file schema alongside expected schema. + */ +private class FileSchemaAccessors + extends SchemaWithPartnerVisitor.PartnerAccessors[Type] { + + override def fieldPartner(partnerStruct: Type, fieldId: Int, name: String): Type = { + if (partnerStruct == null) return null + val structType = partnerStruct.asStructType() + val field = structType.field(fieldId) + if (field == null) null else field.`type`() + } + + override def listElementPartner(partnerList: Type): Type = { + if (partnerList == null) return null + partnerList.asListType().elementType() + } + + override def mapKeyPartner(partnerMap: Type): Type = { + if (partnerMap == null) return null + partnerMap.asMapType().keyType() + } + + override def mapValuePartner(partnerMap: Type): Type = { + if (partnerMap == null) return null + partnerMap.asMapType().valueType() + } +} diff --git a/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuIcebergParquetReaderUtils.scala b/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuIcebergParquetReaderUtils.scala new file mode 100644 index 00000000000..4dbede5a5d3 --- /dev/null +++ b/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuIcebergParquetReaderUtils.scala @@ -0,0 +1,72 @@ +/* + * Copyright (c) 2025-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 org.apache.hadoop.conf.Configuration +import org.apache.iceberg.MetadataColumns +import org.apache.iceberg.hadoop.HadoopInputFile +import org.apache.iceberg.io.InputFile +import org.apache.iceberg.shaded.org.apache.parquet.{HadoopReadOptions, ParquetReadOptions} +import org.apache.iceberg.shaded.org.apache.parquet.schema.{ + MessageType => ShadedMessageType, Types => ShadedTypes} +import org.apache.iceberg.shaded.org.apache.parquet.schema.PrimitiveType.{ + PrimitiveTypeName => ShadedPrimitiveTypeName} +import org.apache.iceberg.shaded.org.apache.parquet.schema.Type.{Repetition => ShadedRepetition} + +object GpuIcebergParquetReaderUtils { + private val READ_PROPERTIES_TO_REMOVE = Set( + "parquet.read.filter", + "parquet.private.read.filter.predicate", + "parquet.read.support.class") + + /** + * Adds the leading file-global row index emitted by the cuDF deletion-vector reader to the + * schema consumed by the Iceberg post-processor. + */ + private[iceberg] def withNativeRowIndex( + fileReadSchema: ShadedMessageType): ShadedMessageType = { + val rowPosition = ShadedTypes + .primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.REQUIRED) + .id(MetadataColumns.ROW_POSITION.fieldId()) + .named(MetadataColumns.ROW_POSITION.name()) + new ShadedMessageType( + fileReadSchema.getName, + (rowPosition +: fileReadSchema.getFields.asScala).asJava) + } + + def buildReaderOptions(file: InputFile, split: Option[(Long, Long)]) + : ParquetReadOptions = { + var optionsBuilder: ParquetReadOptions.Builder = null + file match { + case hadoop: HadoopInputFile => + // remove read properties already set that may conflict with this read + val conf = new Configuration(hadoop.getConf) + for (property <- READ_PROPERTIES_TO_REMOVE) { + conf.unset(property) + } + optionsBuilder = HadoopReadOptions.builder(conf) + case _ => + optionsBuilder = ParquetReadOptions.builder() + } + split.foreach { case (start, length) => + optionsBuilder = optionsBuilder.withRange(start, start + length) + } + optionsBuilder.build + } +} diff --git a/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ThreadConf.scala b/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ThreadConf.scala new file mode 100644 index 00000000000..e37e749da49 --- /dev/null +++ b/iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ThreadConf.scala @@ -0,0 +1,21 @@ +/* + * Copyright (c) 2025-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 + +trait ThreadConf + +case object SingleFile extends ThreadConf diff --git a/iceberg-common/src/main/scala/org/apache/iceberg/io/GpuPositionDeleteFieldIds.scala b/iceberg-common/src/main/scala/org/apache/iceberg/io/GpuPositionDeleteFieldIds.scala new file mode 100644 index 00000000000..e60fd853a84 --- /dev/null +++ b/iceberg-common/src/main/scala/org/apache/iceberg/io/GpuPositionDeleteFieldIds.scala @@ -0,0 +1,24 @@ +/* + * Copyright (c) 2025-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 org.apache.iceberg.io + +import org.apache.iceberg.MetadataColumns.{DELETE_FILE_PATH, DELETE_FILE_POS} + +private[io] object GpuPositionDeleteFieldIds { + val FILE_AND_POS_FIELD_IDS: Set[Integer] = Set( + DELETE_FILE_PATH.fieldId(), DELETE_FILE_POS.fieldId()) +} diff --git a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/data/GpuDeleteFilter.scala b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/data/GpuDeleteFilter.scala index 47c810a4b2e..0027f1f134f 100644 --- a/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/data/GpuDeleteFilter.scala +++ b/iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/data/GpuDeleteFilter.scala @@ -24,7 +24,7 @@ import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource} import com.nvidia.spark.rapids.GpuMetric.{JOIN_TIME, OP_TIME_LEGACY} import com.nvidia.spark.rapids.fileio.iceberg.{IcebergFileIO, IcebergInputFile} import com.nvidia.spark.rapids.iceberg.ShimUtils -import com.nvidia.spark.rapids.iceberg.data.GpuDeleteFilter2._ +import com.nvidia.spark.rapids.iceberg.data.GpuDeleteFileInfo._ import com.nvidia.spark.rapids.iceberg.fieldIndex import com.nvidia.spark.rapids.iceberg.parquet.GpuIcebergParquetReaderConf import org.apache.iceberg.{DeleteFile, FileContent, MetadataColumns, Schema} @@ -72,6 +72,68 @@ object GpuDeleteFileInfo { val effectivePositionDeletes = if (deletionVector.isDefined) Seq.empty else positionDeletes new GpuDeleteFileInfo(deletionVector, equalityDeletes.toSeq ++ effectivePositionDeletes) } + + private[iceberg] val DELETE_EXTRA_METADATA_COLUMNS: Seq[NestedField] = Seq( + MetadataColumns.FILE_PATH, + MetadataColumns.ROW_POSITION) + + private[iceberg] val DELETE_EXTRA_METADATA_COLUMN_IDS: Set[Int] = + DELETE_EXTRA_METADATA_COLUMNS + .map(_.fieldId()) + .toSet + + private[iceberg] val POS_DELETE_SCHEMA: Schema = new Schema( + MetadataColumns.DELETE_FILE_PATH, + MetadataColumns.DELETE_FILE_POS) + + + private[iceberg] def mergeColumn( + batch: ColumnarBatch, srcColIdx: Int, destColIdx: Int) + (mergeOp: (GpuColumnVector, GpuColumnVector) => GpuColumnVector): ColumnarBatch = { + require(srcColIdx >= 0 && srcColIdx < batch.numCols(), + s"Invalid src column index: $srcColIdx, numCols: ${batch.numCols()}") + require(destColIdx >= 0 && destColIdx < batch.numCols(), + s"Invalid dest column index: $destColIdx, numCols: ${batch.numCols()}") + require(srcColIdx != destColIdx, "srcColIdx and destColIdx should be different") + + val srcVec = batch.column(srcColIdx).asInstanceOf[GpuColumnVector] + val destVec = batch.column(destColIdx).asInstanceOf[GpuColumnVector] + + withResource(batch) { _ => + closeOnExcept(mergeOp(srcVec, destVec)) { mergeVec => + val newColumns = new Array[ColumnVector](batch.numCols() - 1) + for (i <- 0 until batch.numCols() - 1) { + if (i == destColIdx) { + newColumns(i) = mergeVec + } else { + newColumns(i) = batch.column(i).asInstanceOf[GpuColumnVector].incRefCount() + } + } + new ColumnarBatch(newColumns, batch.numRows()) + } + } + } + + private[iceberg] def filterAndDrop(batch: ColumnarBatch, + isDeletedColIdx: Int, + outputDataType: Array[DataType], + dropMask: Array[Boolean] = Array.empty): ColumnarBatch = { + withResource(batch) { _ => + withResource(GpuColumnVector.from(batch)) { table => + withResource(table.getColumn(isDeletedColIdx).not()) { maskCv => + withResource(table.filter(maskCv)) { newTable => + if (dropMask.nonEmpty) { + withResource(GpuColumnVector.from(newTable, outputDataType)) { newBatch => + GpuColumnVector.dropColumns(newBatch, dropMask) + } + } else { + GpuColumnVector.from(newTable, outputDataType) + } + } + } + } + } + } } class GpuDeleteFilter( @@ -361,70 +423,6 @@ class GpuDeleteFilter( } } -object GpuDeleteFilter2 { - private[iceberg] val DELETE_EXTRA_METADATA_COLUMNS: Seq[NestedField] = Seq( - MetadataColumns.FILE_PATH, - MetadataColumns.ROW_POSITION) - - private[iceberg] val DELETE_EXTRA_METADATA_COLUMN_IDS: Set[Int] = - DELETE_EXTRA_METADATA_COLUMNS - .map(_.fieldId()) - .toSet - - private[iceberg] val POS_DELETE_SCHEMA: Schema = new Schema( - MetadataColumns.DELETE_FILE_PATH, - MetadataColumns.DELETE_FILE_POS) - - - private[iceberg] def mergeColumn( - batch: ColumnarBatch, srcColIdx: Int, destColIdx: Int) - (mergeOp: (GpuColumnVector, GpuColumnVector) => GpuColumnVector): ColumnarBatch = { - require(srcColIdx >= 0 && srcColIdx < batch.numCols(), - s"Invalid src column index: $srcColIdx, numCols: ${batch.numCols()}") - require(destColIdx >= 0 && destColIdx < batch.numCols(), - s"Invalid dest column index: $destColIdx, numCols: ${batch.numCols()}") - require(srcColIdx != destColIdx, "srcColIdx and destColIdx should be different") - - val srcVec = batch.column(srcColIdx).asInstanceOf[GpuColumnVector] - val destVec = batch.column(destColIdx).asInstanceOf[GpuColumnVector] - - withResource(batch) { _ => - closeOnExcept(mergeOp(srcVec, destVec)) { mergeVec => - val newColumns = new Array[ColumnVector](batch.numCols() - 1) - for (i <- 0 until batch.numCols() - 1) { - if (i == destColIdx) { - newColumns(i) = mergeVec - } else { - newColumns(i) = batch.column(i).asInstanceOf[GpuColumnVector].incRefCount() - } - } - new ColumnarBatch(newColumns, batch.numRows()) - } - } - } - - private[iceberg] def filterAndDrop(batch: ColumnarBatch, - isDeletedColIdx: Int, - outputDataType: Array[DataType], - dropMask: Array[Boolean] = Array.empty): ColumnarBatch = { - withResource(batch) { _ => - withResource(GpuColumnVector.from(batch)) { table => - withResource(table.getColumn(isDeletedColIdx).not()) { maskCv => - withResource(table.filter(maskCv)) { newTable => - if (dropMask.nonEmpty) { - withResource(GpuColumnVector.from(newTable, outputDataType)) { newBatch => - GpuColumnVector.dropColumns(newBatch, dropMask) - } - } else { - GpuColumnVector.from(newTable, outputDataType) - } - } - } - } - } - } -} - private case class DeleteFilterContext( buildBatch: LazySpillableColumnarBatch, buildKeys: Seq[GpuExpression], 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 0b11e557a87..73c8f023e29 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 @@ -574,35 +574,6 @@ private class ActionBuildingVisitor( } } -/** - * Partner accessors to navigate file schema alongside expected schema. - */ -private class FileSchemaAccessors - extends SchemaWithPartnerVisitor.PartnerAccessors[Type] { - - override def fieldPartner(partnerStruct: Type, fieldId: Int, name: String): Type = { - if (partnerStruct == null) return null - val structType = partnerStruct.asStructType() - val field = structType.field(fieldId) - if (field == null) null else field.`type`() - } - - override def listElementPartner(partnerList: Type): Type = { - if (partnerList == null) return null - partnerList.asListType().elementType() - } - - override def mapKeyPartner(partnerMap: Type): Type = { - if (partnerMap == null) return null - partnerMap.asMapType().keyType() - } - - override def mapValuePartner(partnerMap: Type): Type = { - if (partnerMap == null) return null - partnerMap.asMapType().valueType() - } -} - /** Processes columnar batch after reading from parquet file. * * Apache iceberg uses a lazy approach to deal with schema evolution, e.g. when you 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 6a080eaf8d9..85255637d63 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 @@ -33,18 +33,12 @@ 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.hadoop.HadoopInputFile -import org.apache.iceberg.io.InputFile import org.apache.iceberg.mapping.NameMapping import org.apache.iceberg.parquet._ -import org.apache.iceberg.shaded.org.apache.parquet.{HadoopReadOptions, ParquetReadOptions} +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, Types => ShadedTypes} -import org.apache.iceberg.shaded.org.apache.parquet.schema.PrimitiveType.{ - PrimitiveTypeName => ShadedPrimitiveTypeName} -import org.apache.iceberg.shaded.org.apache.parquet.schema.Type.{Repetition => ShadedRepetition} +import org.apache.iceberg.shaded.org.apache.parquet.schema.{MessageType => ShadedMessageType} import org.apache.parquet.hadoop.metadata.BlockMetaData import org.apache.spark.internal.Logging @@ -63,7 +57,7 @@ case class IcebergPartitionedFile( lazy val path: Path = new Path(new URI(urlEncodedPath)) def parquetReadOptions: ParquetReadOptions = { - GpuIcebergParquetReader.buildReaderOptions(file.getDelegate, split) + GpuIcebergParquetReaderUtils.buildReaderOptions(file.getDelegate, split) } def newReader(metrics: Map[String, GpuMetric] = Map.empty): ParquetFileReader = { @@ -115,10 +109,6 @@ case class IcebergPartitionedFile( } } -sealed trait ThreadConf - -case object SingleFile extends ThreadConf - case class MultiThread( poolConfBuilder: ThreadPoolConfBuilder, maxNumFilesProcessed: Int, @@ -298,7 +288,7 @@ trait GpuIcebergParquetReader extends Iterator[ColumnarBatch] with AutoCloseable ) val postProcessorReadSchema = if (hasDeletionVector) { - GpuIcebergParquetReader.withNativeRowIndex(fileReadSchema) + GpuIcebergParquetReaderUtils.withNativeRowIndex(fileReadSchema) } else { fileReadSchema } @@ -306,45 +296,3 @@ trait GpuIcebergParquetReader extends Iterator[ColumnarBatch] with AutoCloseable } } } - -object GpuIcebergParquetReader { - private val READ_PROPERTIES_TO_REMOVE = Set( - "parquet.read.filter", - "parquet.private.read.filter.predicate", - "parquet.read.support.class") - - /** - * Adds the leading file-global row index emitted by the cuDF deletion-vector reader to the - * schema consumed by the Iceberg post-processor. - */ - private[iceberg] def withNativeRowIndex( - fileReadSchema: ShadedMessageType): ShadedMessageType = { - val rowPosition = ShadedTypes - .primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.REQUIRED) - .id(MetadataColumns.ROW_POSITION.fieldId()) - .named(MetadataColumns.ROW_POSITION.name()) - new ShadedMessageType( - fileReadSchema.getName, - (rowPosition +: fileReadSchema.getFields.asScala).asJava) - } - - def buildReaderOptions(file: InputFile, split: Option[(Long, Long)]) - : ParquetReadOptions = { - var optionsBuilder: ParquetReadOptions.Builder = null - file match { - case hadoop: HadoopInputFile => - // remove read properties already set that may conflict with this read - val conf = new Configuration(hadoop.getConf) - for (property <- READ_PROPERTIES_TO_REMOVE) { - conf.unset(property) - } - optionsBuilder = HadoopReadOptions.builder(conf) - case _ => - optionsBuilder = ParquetReadOptions.builder() - } - split.foreach { case (start, length) => - optionsBuilder = optionsBuilder.withRange(start, start + length) - } - optionsBuilder.build - } -} diff --git a/iceberg/common/src/main/scala/org/apache/iceberg/io/rolling.scala b/iceberg/common/src/main/scala/org/apache/iceberg/io/rolling.scala index f7904acda88..d59bd475959 100644 --- a/iceberg/common/src/main/scala/org/apache/iceberg/io/rolling.scala +++ b/iceberg/common/src/main/scala/org/apache/iceberg/io/rolling.scala @@ -1,5 +1,5 @@ /* - * Copyright (c) 2025, NVIDIA CORPORATION. + * Copyright (c) 2025-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. @@ -24,9 +24,8 @@ import scala.collection.JavaConverters._ import com.nvidia.spark.rapids.{GpuColumnVector, SpillableColumnarBatch} import com.nvidia.spark.rapids.Arm.withResource import org.apache.iceberg._ -import org.apache.iceberg.MetadataColumns.{DELETE_FILE_PATH, DELETE_FILE_POS} import org.apache.iceberg.encryption.EncryptedOutputFile -import org.apache.iceberg.io.GpuPositionDeleteFileWriter.FILE_AND_POS_FIELD_IDS +import org.apache.iceberg.io.GpuPositionDeleteFieldIds.FILE_AND_POS_FIELD_IDS import org.apache.iceberg.relocated.com.google.common.collect.Lists import org.apache.iceberg.spark.source.GpuSparkFileWriterFactory import org.apache.iceberg.util.CharSequenceSet @@ -231,9 +230,3 @@ class GpuPositionDeleteFileWriter( } } } - -object GpuPositionDeleteFileWriter { - private[io] val FILE_AND_POS_FIELD_IDS: Set[Integer] = Set( - DELETE_FILE_PATH.fieldId(), DELETE_FILE_POS.fieldId()) -} - diff --git a/scala2.13/iceberg-common/pom.xml b/scala2.13/iceberg-common/pom.xml index 8e4ee7abd71..0390c3fc5d7 100644 --- a/scala2.13/iceberg-common/pom.xml +++ b/scala2.13/iceberg-common/pom.xml @@ -40,6 +40,11 @@ + + com.nvidia + spark-rapids-jni + ${jni.classifier} + org.apache.iceberg iceberg-spark-runtime-${iceberg.artifact.suffix}_${scala.binary.version} 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 ab5a7390c13..1e1e567b933 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 @@ -1024,7 +1024,7 @@ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll { parquetInfo, new JHashMap[Integer, Any](), expectedSchema, - GpuIcebergParquetReader.withNativeRowIndex(shadedSchema), + GpuIcebergParquetReaderUtils.withNativeRowIndex(shadedSchema), Map.empty) assert(!processor.displayActionPlan().contains("FetchRowPosition")) @@ -1063,7 +1063,7 @@ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll { parquetInfo, new JHashMap[Integer, Any](), expectedSchema, - GpuIcebergParquetReader.withNativeRowIndex(shadedSchema), + GpuIcebergParquetReaderUtils.withNativeRowIndex(shadedSchema), Map.empty) assert(processor.displayActionPlan() ==