From 3725ab56e93007243b916cfbf263727d0ee9d8fd Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Mon, 31 Aug 2026 13:00:03 -0700 Subject: [PATCH 1/7] Move independent Iceberg helpers to root-safe module Signed-off-by: Gera Shegalov --- dist/unshimmed-common-from-single-shim.txt | 9 -- iceberg-common/pom.xml | 5 + .../rapids/iceberg/IcebergDeletionVector.java | 3 +- .../iceberg/spark/RapidsSparkCatalog.java | 0 .../spark/RapidsSparkSessionCatalog.java | 0 .../spark/source/RapidsSparkTable.java | 0 .../GpuIcebergParquetReaderUtils.scala | 72 ++++++++++ .../rapids/iceberg/parquet/ThreadConf.scala | 21 +++ .../io/GpuPositionDeleteFieldIds.scala | 24 ++++ .../rapids/iceberg/data/GpuDeleteFilter.scala | 128 +++++++++--------- .../spark/rapids/iceberg/parquet/reader.scala | 60 +------- .../scala/org/apache/iceberg/io/rolling.scala | 11 +- scala2.13/iceberg-common/pom.xml | 5 + .../iceberg/GpuPostProcessorSuite.scala | 4 +- 14 files changed, 200 insertions(+), 142 deletions(-) rename {iceberg/common => iceberg-common}/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergDeletionVector.java (98%) rename {iceberg/common => iceberg-common}/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java (100%) rename {iceberg/common => iceberg-common}/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java (100%) rename {iceberg/common => iceberg-common}/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java (100%) create mode 100644 iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuIcebergParquetReaderUtils.scala create mode 100644 iceberg-common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/ThreadConf.scala create mode 100644 iceberg-common/src/main/scala/org/apache/iceberg/io/GpuPositionDeleteFieldIds.scala diff --git a/dist/unshimmed-common-from-single-shim.txt b/dist/unshimmed-common-from-single-shim.txt index 54ce2904976..3789a773dfe 100644 --- a/dist/unshimmed-common-from-single-shim.txt +++ b/dist/unshimmed-common-from-single-shim.txt @@ -14,22 +14,13 @@ 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 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/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/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java b/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java similarity index 100% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java rename to iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java 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/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() == From 3ef9f1d64b4579406e703a3baf8b166b1a732ec8 Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Mon, 31 Aug 2026 15:19:41 -0700 Subject: [PATCH 2/7] Keep Iceberg catalog wrappers runtime-specific Signed-off-by: Gera Shegalov --- dist/unshimmed-common-from-single-shim.txt | 3 +++ .../nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java | 0 .../spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java | 0 .../spark/rapids/iceberg/spark/source/RapidsSparkTable.java | 0 4 files changed, 3 insertions(+) rename {iceberg-common => iceberg/common}/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java (100%) rename {iceberg-common => iceberg/common}/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java (100%) rename {iceberg-common => iceberg/common}/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java (100%) diff --git a/dist/unshimmed-common-from-single-shim.txt b/dist/unshimmed-common-from-single-shim.txt index 3789a773dfe..d3f7a5a6892 100644 --- a/dist/unshimmed-common-from-single-shim.txt +++ b/dist/unshimmed-common-from-single-shim.txt @@ -17,6 +17,9 @@ com/nvidia/spark/rapids/iceberg/GpuInternalRowBase.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/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 diff --git a/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java b/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java similarity index 100% rename from iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java rename to iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java diff --git a/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java b/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java similarity index 100% rename from iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java rename to iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java diff --git a/iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java b/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java similarity index 100% rename from iceberg-common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java rename to iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java From 230ec0b7ed45df5e9e04357d49bbd2193db72fe2 Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Wed, 2 Sep 2026 10:51:00 -0700 Subject: [PATCH 3/7] Update row-lineage post-processor tests Signed-off-by: Gera Shegalov --- .../nvidia/spark/rapids/iceberg/GpuPostProcessorSuite.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 ebfa550c8dc..b44a1606c55 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 @@ -1178,7 +1178,7 @@ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll { parquetInfo, new JHashMap[Integer, Any](), expectedSchema, - GpuIcebergParquetReader.withNativeRowIndex(shadedSchema), + GpuIcebergParquetReaderUtils.withNativeRowIndex(shadedSchema), Map.empty) assert(processor.displayActionPlan() == @@ -1215,7 +1215,7 @@ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll { parquetInfo, constants, expectedSchema, - GpuIcebergParquetReader.withNativeRowIndex(shadedSchema), + GpuIcebergParquetReaderUtils.withNativeRowIndex(shadedSchema), Map.empty) val rowPositions = closeOnExcept(CudfColumnVector.fromLongs(0L, 2L, 5L)) { column => From f131577eb56709310587155f40215f8463fe8fe3 Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Tue, 8 Sep 2026 22:11:58 -0700 Subject: [PATCH 4/7] Fix Iceberg reader utility test after up-merge Signed-off-by: Gera Shegalov --- .../com/nvidia/spark/rapids/iceberg/GpuPostProcessorSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 01aa3f62ccf..e0a293999bc 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 @@ -1812,7 +1812,7 @@ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll { val metric = new LocalGpuMetric // The _pos column the DV path injects is dropped, so it must not count as decoded. val processor = new GpuParquetReaderPostProcessor(info, new JHashMap[Integer, Any](), - expectedSchema, GpuIcebergParquetReader.withNativeRowIndex(shaded), + expectedSchema, GpuIcebergParquetReaderUtils.withNativeRowIndex(shaded), Map(GPU_OUTPUT_BATCH_BYTES -> metric)) val withPos = StructType(Array( StructField("_pos", LongType, true), StructField("long_col", LongType, true))) From 8806121f5c397240502ef326451990bcb1f0bfce Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Wed, 9 Sep 2026 15:54:04 -0700 Subject: [PATCH 5/7] Finalize Iceberg root versus shim placement Signed-off-by: Gera Shegalov --- dist/build/package-parallel-worlds.py | 12 +++++++++-- dist/keep-in-spark-shim-dirs.txt | 9 ++++++++ dist/scripts/binary-dedupe.sh | 6 ++++++ dist/scripts/build-unshim-parallel-world.py | 12 +++++++++-- dist/unshimmed-common-from-single-shim.txt | 7 ------- iceberg-common/pom.xml | 8 +++++++ .../iceberg/spark/RapidsSparkCatalog.java | 21 +++++++++++++++++++ .../spark/RapidsSparkSessionCatalog.java | 21 +++++++++++++++++++ .../spark/source/RapidsSparkTable.java | 21 +++++++++++++++++++ .../iceberg/spark/source/GpuBaseReader.java | 7 +++++++ .../spark/source/GpuSparkPlanningUtil.java | 21 +++++++++++++++++++ scala2.13/iceberg-common/pom.xml | 8 +++++++ 12 files changed, 142 insertions(+), 11 deletions(-) rename {iceberg/common/src/main => iceberg-common/src/main/spark350}/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java (84%) rename {iceberg/common/src/main => iceberg-common/src/main/spark350}/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java (87%) rename {iceberg/common/src/main => iceberg-common/src/main/spark350}/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java (97%) rename {iceberg/iceberg-1-6-x/src/main => iceberg-common/src/main/spark350}/java/org/apache/iceberg/spark/source/GpuBaseReader.java (86%) rename {iceberg/common/src/main => iceberg-common/src/main/spark350}/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java (75%) diff --git a/dist/build/package-parallel-worlds.py b/dist/build/package-parallel-worlds.py index ef279e25f4c..f8ca6738f80 100644 --- a/dist/build/package-parallel-worlds.py +++ b/dist/build/package-parallel-worlds.py @@ -204,6 +204,10 @@ def root_safe_module_class_members(classifier): root_safe_modules = read_patterns(os.sep.join([dist_dir, 'root-safe-module-classes.txt'])) from_single_shim_or_each = from_single_shim + from_each iceberg_audit_runtimes = {} +# Classifiers are processed newest-first. Let older classifiers contribute +# conditional root-safe classes that are absent from newer classifiers, without +# overwriting a newer implementation of the same class path. +promoted_root_safe_members = set() for bv in buildver_list: classifier = 'spark' + bv @@ -225,7 +229,7 @@ def root_safe_module_class_members(classifier): # IMPORTANT unconditional extract from the highest Spark version to the top if bv == buildver_list[0] and art == 'sql-plugin-api': zip_handle.extractall(path=top_dist_jar_dir) - if bv == buildver_list[0] and art == 'aggregator': + if art == 'aggregator': namelist = zip_handle.namelist() namelist_set = set(namelist) root_safe_members = root_safe_module_class_members(classifier) @@ -234,9 +238,13 @@ def root_safe_module_class_members(classifier): raise Exception( "root-safe module classes missing from aggregator: %s" % ", ".join(missing_members)) + new_root_safe_members = ( + root_safe_members - promoted_root_safe_members) zip_handle.extractall( path=top_dist_jar_dir, - members=[name for name in namelist if name in root_safe_members]) + members=[name for name in namelist + if name in new_root_safe_members]) + promoted_root_safe_members.update(root_safe_members) # TODO deprecate namelist = zip_handle.namelist() glob_list = from_single_shim_or_each if bv == buildver_list[0] else from_each diff --git a/dist/keep-in-spark-shim-dirs.txt b/dist/keep-in-spark-shim-dirs.txt index 41dded9f0ff..46d220ed661 100644 --- a/dist/keep-in-spark-shim-dirs.txt +++ b/dist/keep-in-spark-shim-dirs.txt @@ -53,3 +53,12 @@ org/apache/spark/sql/rapids/execution/GpuBroadcastExchangeExec*.class # exchange execs. Keep the rule and helpers in the selected shim loader. com/nvidia/spark/rapids/GpuTransitionOverrides.class com/nvidia/spark/rapids/GpuTransitionOverrides$$*.class + +# Iceberg implementation classes compile against Spark- and Iceberg-version-specific +# APIs. Root-selected Iceberg classes are excluded by binary-dedupe.sh; keep every +# other Iceberg class in the selected shim loader instead of spark-shared. +com/nvidia/spark/rapids/*Iceberg*.class +com/nvidia/spark/rapids/fileio/iceberg/*.class +com/nvidia/spark/rapids/iceberg/*.class +org/apache/iceberg/*.class +org/apache/spark/sql/rapids/*Iceberg*.class diff --git a/dist/scripts/binary-dedupe.sh b/dist/scripts/binary-dedupe.sh index a5ac3346c79..7ae2fa62a7c 100755 --- a/dist/scripts/binary-dedupe.sh +++ b/dist/scripts/binary-dedupe.sh @@ -182,6 +182,12 @@ function filter_keep_in_spark_shim_dirs() { path_without_leading_slash="${class_resource#/}" class_file="${path_without_leading_slash#*/}" if keep_in_spark_shim_dirs "$class_file"; then + # Root-layout classes were explicitly selected by the packager and must + # remain eligible for de-duplication so the later root promotion can + # remove their shim copies. + if [[ -f "./parallel-world/$class_file" ]]; then + echo "$class_resource" + fi continue fi echo "$class_resource" diff --git a/dist/scripts/build-unshim-parallel-world.py b/dist/scripts/build-unshim-parallel-world.py index 05e47882618..8ca95a920c0 100644 --- a/dist/scripts/build-unshim-parallel-world.py +++ b/dist/scripts/build-unshim-parallel-world.py @@ -202,6 +202,10 @@ def copy_and_extract_jars( cache_root = target_dir / "unshim-parallel-world-cache" sorted_buildvers = sorted(buildvers, reverse=True) root_buildver = sorted_buildvers[0] + # Classifiers are processed newest-first. Let older classifiers contribute + # conditional root-safe classes that are absent from newer classifiers, + # without overwriting a newer implementation of the same class path. + promoted_root_safe_members = set() for buildver in sorted_buildvers: classifier = "spark%s" % buildver @@ -216,7 +220,7 @@ def copy_and_extract_jars( link_tree_contents(contents_dir, parallel_world / classifier) if buildver == root_buildver and artifact == "sql-plugin-api": link_tree_contents(contents_dir, parallel_world) - if buildver == root_buildver and artifact == "aggregator": + if artifact == "aggregator": root_safe_members = root_safe_module_class_members( base_dir, scala_binary_version, @@ -228,7 +232,11 @@ def copy_and_extract_jars( raise RuntimeError( "root-safe module classes missing from aggregator: %s" % ", ".join(missing_members)) - link_members(contents_dir, parallel_world, sorted(root_safe_members)) + new_root_safe_members = ( + root_safe_members - promoted_root_safe_members) + link_members(contents_dir, parallel_world, + sorted(new_root_safe_members)) + promoted_root_safe_members.update(root_safe_members) patterns = from_each if buildver == root_buildver: diff --git a/dist/unshimmed-common-from-single-shim.txt b/dist/unshimmed-common-from-single-shim.txt index cdc418a2ce5..4faa1243144 100644 --- a/dist/unshimmed-common-from-single-shim.txt +++ b/dist/unshimmed-common-from-single-shim.txt @@ -11,13 +11,6 @@ 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/iceberg/parquet/converter/FromIcebergShaded*.class -com/nvidia/spark/rapids/iceberg/parquet/converter/ToIcebergShaded*.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/spark/source/GpuBaseReader.class -org/apache/iceberg/spark/source/GpuSparkPlanningUtil.class org/apache/spark/sql/rapids/AdaptiveSparkPlanHelperShim* org/apache/spark/sql/rapids/ExecutionPlanCaptureCallback* rapids/*.py diff --git a/iceberg-common/pom.xml b/iceberg-common/pom.xml index dcd5bdfabd9..974e0a05676 100644 --- a/iceberg-common/pom.xml +++ b/iceberg-common/pom.xml @@ -37,6 +37,14 @@ package false true + + -Xlint:all,-serial,-path,-try,-processing,-options,-classfile|-Werror diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java similarity index 84% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java rename to iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java index f86ecf60eee..3fa7c6f1aac 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java +++ b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java @@ -14,6 +14,27 @@ * limitations under the License. */ +/*** spark-rapids-shim-json-lines +{"spark": "350"} +{"spark": "351"} +{"spark": "352"} +{"spark": "353"} +{"spark": "354"} +{"spark": "355"} +{"spark": "356"} +{"spark": "357"} +{"spark": "358"} +{"spark": "359"} +{"spark": "400"} +{"spark": "401"} +{"spark": "402"} +{"spark": "403"} +{"spark": "404"} +{"spark": "411"} +{"spark": "412"} +{"spark": "413"} +spark-rapids-shim-json-lines ***/ + package com.nvidia.spark.rapids.iceberg.spark; import org.apache.iceberg.spark.SparkCatalog; diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java similarity index 87% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java rename to iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java index 1620ba84a75..5e60dcd20bb 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java +++ b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java @@ -14,6 +14,27 @@ * limitations under the License. */ +/*** spark-rapids-shim-json-lines +{"spark": "350"} +{"spark": "351"} +{"spark": "352"} +{"spark": "353"} +{"spark": "354"} +{"spark": "355"} +{"spark": "356"} +{"spark": "357"} +{"spark": "358"} +{"spark": "359"} +{"spark": "400"} +{"spark": "401"} +{"spark": "402"} +{"spark": "403"} +{"spark": "404"} +{"spark": "411"} +{"spark": "412"} +{"spark": "413"} +spark-rapids-shim-json-lines ***/ + package com.nvidia.spark.rapids.iceberg.spark; import com.nvidia.spark.rapids.iceberg.spark.source.RapidsSparkTable; diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java similarity index 97% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java rename to iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java index a631fcee427..0b1128f8db3 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java +++ b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java @@ -14,6 +14,27 @@ * limitations under the License. */ +/*** spark-rapids-shim-json-lines +{"spark": "350"} +{"spark": "351"} +{"spark": "352"} +{"spark": "353"} +{"spark": "354"} +{"spark": "355"} +{"spark": "356"} +{"spark": "357"} +{"spark": "358"} +{"spark": "359"} +{"spark": "400"} +{"spark": "401"} +{"spark": "402"} +{"spark": "403"} +{"spark": "404"} +{"spark": "411"} +{"spark": "412"} +{"spark": "413"} +spark-rapids-shim-json-lines ***/ + package com.nvidia.spark.rapids.iceberg.spark.source; import java.util.Arrays; diff --git a/iceberg/iceberg-1-6-x/src/main/java/org/apache/iceberg/spark/source/GpuBaseReader.java b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java similarity index 86% rename from iceberg/iceberg-1-6-x/src/main/java/org/apache/iceberg/spark/source/GpuBaseReader.java rename to iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java index aa76cc468b5..cf62a437f47 100644 --- a/iceberg/iceberg-1-6-x/src/main/java/org/apache/iceberg/spark/source/GpuBaseReader.java +++ b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java @@ -14,6 +14,13 @@ * limitations under the License. */ +/*** spark-rapids-shim-json-lines +{"spark": "350"} +{"spark": "351"} +{"spark": "352"} +{"spark": "353"} +spark-rapids-shim-json-lines ***/ + package org.apache.iceberg.spark.source; import org.apache.iceberg.types.Type; diff --git a/iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java similarity index 75% rename from iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java rename to iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java index e26cb01584b..4defa3030de 100644 --- a/iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java +++ b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java @@ -14,6 +14,27 @@ * limitations under the License. */ +/*** spark-rapids-shim-json-lines +{"spark": "350"} +{"spark": "351"} +{"spark": "352"} +{"spark": "353"} +{"spark": "354"} +{"spark": "355"} +{"spark": "356"} +{"spark": "357"} +{"spark": "358"} +{"spark": "359"} +{"spark": "400"} +{"spark": "401"} +{"spark": "402"} +{"spark": "403"} +{"spark": "404"} +{"spark": "411"} +{"spark": "412"} +{"spark": "413"} +spark-rapids-shim-json-lines ***/ + package org.apache.iceberg.spark.source; import org.apache.iceberg.ScanTaskGroup; diff --git a/scala2.13/iceberg-common/pom.xml b/scala2.13/iceberg-common/pom.xml index 7f513b3a15c..ae8bd419cdd 100644 --- a/scala2.13/iceberg-common/pom.xml +++ b/scala2.13/iceberg-common/pom.xml @@ -37,6 +37,14 @@ package false true + + -Xlint:all,-serial,-path,-try,-processing,-options,-classfile|-Werror From ac31e5c387a27a0bfa2b4d991e1b49d1cd7492ef Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Wed, 9 Sep 2026 17:25:23 -0700 Subject: [PATCH 6/7] Update Iceberg helper copyright years Signed-off-by: Gera Shegalov --- .../java/org/apache/iceberg/spark/source/GpuBaseReader.java | 2 +- .../org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java index cf62a437f47..9171ff8c57c 100644 --- a/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java +++ b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java @@ -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. diff --git a/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java index 4defa3030de..4a3f453e6cd 100644 --- a/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java +++ b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java @@ -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. From 31a21327eeef0f66672d36a1cb8489960f828e0e Mon Sep 17 00:00:00 2001 From: Gera Shegalov Date: Fri, 11 Sep 2026 11:05:10 -0700 Subject: [PATCH 7/7] Add root-safe provider selection regression test Signed-off-by: Gera Shegalov --- .../test_root_safe_provider_selection.py | 216 ++++++++++++++++++ pom.xml | 5 + scala2.13/pom.xml | 5 + 3 files changed, 226 insertions(+) create mode 100644 dist/scripts/tests/test_root_safe_provider_selection.py diff --git a/dist/scripts/tests/test_root_safe_provider_selection.py b/dist/scripts/tests/test_root_safe_provider_selection.py new file mode 100644 index 00000000000..148286fcdea --- /dev/null +++ b/dist/scripts/tests/test_root_safe_provider_selection.py @@ -0,0 +1,216 @@ +# 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. + +import importlib.util +import os +from pathlib import Path +import subprocess +import tempfile +import unittest +import zipfile + + +DIST_DIR = Path(__file__).resolve().parents[2] +STANDARD_ASSEMBLER = DIST_DIR / "build" / "package-parallel-worlds.py" +FAST_ASSEMBLER = DIST_DIR / "scripts" / "build-unshim-parallel-world.py" +BINARY_DEDUPE = DIST_DIR / "scripts" / "binary-dedupe.sh" + +SHARED = "org/apache/iceberg/Shared.class" +NEWER_ONLY = "org/apache/iceberg/NewerOnly.class" +OLDER_ONLY = "org/apache/iceberg/OlderOnly.class" +NEWER_IMPL = "org/apache/iceberg/NewerImpl.class" +OLDER_IMPL = "org/apache/iceberg/OlderImpl.class" + + +def write_jar(path, entries): + path.parent.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(path, "w") as jar: + for name, contents in entries.items(): + jar.writestr(name, contents) + + +def artifact_path(base_dir, artifact, buildver): + artifact_id = "rapids-4-spark-%s_2.13" % artifact + return (base_dir / artifact / "target" / ("spark%s" % buildver) / + ("%s-1.0-spark%s.jar" % (artifact_id, buildver))) + + +def create_artifacts(base_dir): + for buildver in ("353", "413"): + write_jar(artifact_path(base_dir, "sql-plugin-api", buildver), {}) + + write_jar(artifact_path(base_dir, "iceberg-common", "413"), { + SHARED: b"module-shared-413", + NEWER_ONLY: b"module-newer-only", + }) + write_jar(artifact_path(base_dir, "aggregator", "413"), { + SHARED: b"aggregator-shared-413", + NEWER_ONLY: b"aggregator-newer-only", + NEWER_IMPL: b"aggregator-newer-impl", + }) + write_jar(artifact_path(base_dir, "iceberg-common", "353"), { + SHARED: b"module-shared-353", + OLDER_ONLY: b"module-older-only", + }) + write_jar(artifact_path(base_dir, "aggregator", "353"), { + SHARED: b"aggregator-shared-353", + OLDER_ONLY: b"aggregator-older-only", + OLDER_IMPL: b"aggregator-older-impl", + }) + + +def read_bytes(root, entry): + return (root / entry).read_bytes() + + +class FakeAttributes: + def get(self, name): + if name == "artifact_csv": + return "sql-plugin-api,aggregator" + raise KeyError(name) + + +class FakeProject: + def __init__(self, source_dir, project_dir, target_dir, repository_dir): + self.properties = { + "included_buildvers": "353,413", + "spark.rapids.source.basedir": str(source_dir), + "spark.rapids.project.basedir": str(project_dir), + "project.version": "1.0", + "scala.binary.version": "2.13", + "project.build.directory": str(target_dir), + "env.ART_URL": "", + "maven.local.repository": str(repository_dir), + "should.build.conventional.jar": False, + } + + def getProperty(self, name): + return self.properties.get(name) + + +def execfile_compat(path, globals_dict): + with open(path, "rb") as source: + code = compile(source.read(), str(path), "exec") + exec(code, globals_dict) + + +def load_fast_assembler(): + spec = importlib.util.spec_from_file_location("build_unshim_parallel_world", FAST_ASSEMBLER) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class RootSafeProviderSelectionTest(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.root = Path(self.temp_dir.name) + self.project_dir = self.root / "project" + self.source_dir = self.root / "source" + self.config_dir = self.source_dir / "dist" + self.config_dir.mkdir(parents=True) + (self.config_dir / "build").mkdir() + (self.config_dir / "unshimmed-common-from-single-shim.txt").write_text("") + (self.config_dir / "unshimmed-from-each-spark3xx.txt").write_text("") + (self.config_dir / "root-safe-module-classes.txt").write_text("iceberg-common\n") + (self.config_dir / "keep-in-spark-shared.txt").write_text("") + (self.config_dir / "keep-in-spark-shim-dirs.txt").write_text( + "org/apache/iceberg/*.class\n") + (self.config_dir / "build" / "iceberg_runtime.py").write_text( + "def coordinates(zip_handle, buildver, scala_version, get_property):\n" + " return []\n") + create_artifacts(self.project_dir) + + def tearDown(self): + self.temp_dir.cleanup() + + def assemble_standard(self): + target_dir = self.root / "standard-target" + (target_dir / "deps").mkdir(parents=True) + globals_dict = { + "attributes": FakeAttributes(), + "project": FakeProject( + self.source_dir, self.project_dir, target_dir, self.root / "repository"), + "execfile": execfile_compat, + "self": self, + } + execfile_compat(STANDARD_ASSEMBLER, globals_dict) + return target_dir + + def assemble_fast(self): + target_dir = self.root / "fast-target" + fast = load_fast_assembler() + fast.copy_and_extract_jars( + self.project_dir, target_dir, "2.13", "1.0", ["353", "413"], + [], [], ["iceberg-common"]) + return target_dir + + def assert_provider_selection(self, target_dir): + parallel_world = target_dir / "parallel-world" + self.assertEqual(b"aggregator-shared-413", read_bytes(parallel_world, SHARED)) + self.assertEqual(b"aggregator-newer-only", read_bytes(parallel_world, NEWER_ONLY)) + self.assertEqual(b"aggregator-older-only", read_bytes(parallel_world, OLDER_ONLY)) + self.assertFalse((parallel_world / NEWER_IMPL).exists()) + self.assertFalse((parallel_world / OLDER_IMPL).exists()) + self.assertTrue((parallel_world / "spark413" / NEWER_IMPL).is_file()) + self.assertTrue((parallel_world / "spark353" / OLDER_IMPL).is_file()) + + def run_dedupe(self, target_dir): + parallel_world = target_dir / "parallel-world" + # Root-safe classes must remain binary-compatible across the worlds where they occur. + # Normalize the synthetic shared class after verifying which provider supplied the root. + (parallel_world / "spark353" / SHARED).write_bytes( + read_bytes(parallel_world / "spark413", SHARED)) + env = os.environ.copy() + env.update({ + "UNSHIM_FAST": "1", + "UNSHIMMED_COMMON_FROM_SINGLE_SHIM_TXT": str( + self.config_dir / "unshimmed-common-from-single-shim.txt"), + "KEEP_IN_SPARK_SHARED_TXT": str( + self.config_dir / "keep-in-spark-shared.txt"), + "KEEP_IN_SPARK_SHIM_DIRS_TXT": str( + self.config_dir / "keep-in-spark-shim-dirs.txt"), + "UNSHIM_ANALYZER_SCRIPT": str(self.root / "missing-analyzer.py"), + }) + result = subprocess.run([str(BINARY_DEDUPE)], cwd=target_dir, env=env, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True) + self.assertEqual(0, result.returncode, result.stdout + result.stderr) + + def assert_final_layout(self, target_dir): + parallel_world = target_dir / "parallel-world" + for helper in (SHARED, NEWER_ONLY, OLDER_ONLY): + self.assertTrue((parallel_world / helper).is_file()) + self.assertFalse((parallel_world / "spark413" / helper).exists()) + self.assertFalse((parallel_world / "spark353" / helper).exists()) + self.assertTrue((parallel_world / "spark413" / NEWER_IMPL).is_file()) + self.assertTrue((parallel_world / "spark353" / OLDER_IMPL).is_file()) + self.assertFalse((parallel_world / "spark353" / NEWER_IMPL).exists()) + self.assertFalse((parallel_world / "spark413" / OLDER_IMPL).exists()) + shared_iceberg = parallel_world / "spark-shared" / "org/apache/iceberg" + self.assertEqual([], list(shared_iceberg.rglob("*.class"))) + + def test_provider_selection_and_dedupe_for_both_assemblers(self): + for name, assemble in ( + ("standard", self.assemble_standard), + ("fast", self.assemble_fast)): + with self.subTest(assembler=name): + target_dir = assemble() + self.assert_provider_selection(target_dir) + self.run_dedupe(target_dir) + self.assert_final_layout(target_dir) + + +if __name__ == "__main__": + unittest.main() diff --git a/pom.xml b/pom.xml index 8484ed5c513..5d7e55d0ab7 100644 --- a/pom.xml +++ b/pom.xml @@ -2056,6 +2056,11 @@ This will force full Scala code rebuild in downstream modules. + + + + + + + +