Skip to content

Commit 213ce7b

Browse files
liurenjie1024liurenjie1024
committed
Merge remote-tracking branch 'upstream/branch-25.12' into ray/13604
2 parents 3a6e1ba + 6e6fb8f commit 213ce7b

12 files changed

Lines changed: 633 additions & 36 deletions

File tree

‎delta-lake/delta-33x/src/main/scala/com/nvidia/spark/rapids/delta/delta33x/Delta33xProvider.scala‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ import org.apache.hadoop.fs.Path
2525
import org.apache.spark.sql.SparkSession
2626
import org.apache.spark.sql.catalyst.expressions.Expression
2727
import org.apache.spark.sql.connector.catalog.SupportsWrite
28-
import org.apache.spark.sql.delta.{DeltaLog, DeltaParquetFileFormat}
28+
import org.apache.spark.sql.delta.{DeltaDynamicPartitionOverwriteCommand, DeltaLog, DeltaParquetFileFormat}
2929
import org.apache.spark.sql.delta.DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME
3030
import org.apache.spark.sql.delta.catalog.{DeltaCatalog, DeltaTableV2}
3131
import org.apache.spark.sql.delta.commands.{DeleteCommand, MergeIntoCommand, OptimizeTableCommand, UpdateCommand}
@@ -93,7 +93,10 @@ object Delta33xProvider extends DeltaIOProvider {
9393
(a, conf, p, r) => new MergeIntoCommandMeta(a, conf, p, r)),
9494
GpuOverrides.runnableCmd[OptimizeTableCommand](
9595
"Optimize a Delta Lake table",
96-
(a, conf, p, r) => new OptimizeTableCommandMeta(a, conf, p, r))
96+
(a, conf, p, r) => new OptimizeTableCommandMeta(a, conf, p, r)),
97+
GpuOverrides.runnableCmd[DeltaDynamicPartitionOverwriteCommand](
98+
"Dynamic partition overwrite to a Delta Lake table",
99+
(a, conf, p, r) => new DeltaDynamicPartitionOverwriteCommandMeta(a, conf, p, r))
97100
).map(r => (r.getClassFor.asSubclass(classOf[RunnableCommand]), r)).toMap
98101
}
99102

‎delta-lake/delta-33x/src/main/scala/com/nvidia/spark/rapids/delta/delta33x/DeltaDynamicPartitionOverwriteCommandMeta.scala‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
/*
2+
* Copyright (c) 2025, NVIDIA CORPORATION.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.nvidia.spark.rapids.delta.delta33x
18+
19+
import com.nvidia.spark.rapids.{DataFromReplacementRule, RapidsConf, RapidsMeta, RunnableCommandMeta}
20+
import com.nvidia.spark.rapids.delta.RapidsDeltaUtils
21+
22+
import org.apache.spark.sql.GpuDeltaDynamicPartitionOverwriteCommand
23+
import org.apache.spark.sql.delta.DeltaDynamicPartitionOverwriteCommand
24+
import org.apache.spark.sql.delta.commands.DeletionVectorUtils
25+
import org.apache.spark.sql.delta.rapids.GpuDeltaLog
26+
import org.apache.spark.sql.delta.sources.DeltaSQLConf
27+
import org.apache.spark.sql.execution.command.RunnableCommand
28+
29+
class DeltaDynamicPartitionOverwriteCommandMeta(
30+
overwriteCommand: DeltaDynamicPartitionOverwriteCommand,
31+
conf: RapidsConf,
32+
parent: Option[RapidsMeta[_, _, _]],
33+
rule: DataFromReplacementRule)
34+
extends RunnableCommandMeta[DeltaDynamicPartitionOverwriteCommand](overwriteCommand, conf,
35+
parent, rule) {
36+
37+
override def tagSelfForGpu(): Unit = {
38+
if (!conf.isDeltaWriteEnabled) {
39+
willNotWorkOnGpu("Delta Lake output acceleration has been disabled. To enable set " +
40+
s"${RapidsConf.ENABLE_DELTA_WRITE} to true")
41+
}
42+
43+
val dvFeatureEnabled = DeletionVectorUtils.deletionVectorsWritable(
44+
overwriteCommand.deltaTable.deltaLog.unsafeVolatileSnapshot)
45+
46+
if (dvFeatureEnabled && overwriteCommand.conf.getConf(
47+
DeltaSQLConf.DELETE_USE_PERSISTENT_DELETION_VECTORS)) {
48+
// https://github.com/NVIDIA/spark-rapids/issues/8554
49+
willNotWorkOnGpu("Writes with deletion vectors are not supported on GPU")
50+
}
51+
52+
RapidsDeltaUtils.tagForDeltaWrite(this, overwriteCommand.table.schema,
53+
Some(overwriteCommand.deltaTable.deltaLog),
54+
Map.empty, overwriteCommand.deltaTable.spark)
55+
}
56+
57+
override def convertToGpu(): RunnableCommand = {
58+
GpuDeltaDynamicPartitionOverwriteCommand(
59+
new GpuDeltaLog(overwriteCommand.deltaTable.deltaLog, conf),
60+
overwriteCommand.table,
61+
overwriteCommand.deltaTable,
62+
overwriteCommand.query,
63+
overwriteCommand.writeOptions,
64+
overwriteCommand.isByName,
65+
overwriteCommand.analyzedQuery,
66+
)
67+
}
68+
}

‎delta-lake/delta-33x/src/main/scala/org/apache/spark/sql/GpuDeltaDynamicPartitionOverwriteCommand.scala‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Copyright (c) 2025, NVIDIA CORPORATION.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.apache.spark.sql
18+
19+
import org.apache.spark.sql.catalyst.analysis.NamedRelation
20+
import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan, V2WriteCommand}
21+
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
22+
import org.apache.spark.sql.delta.DeltaOptions
23+
import org.apache.spark.sql.delta.catalog.DeltaTableV2
24+
import org.apache.spark.sql.delta.commands.WriteIntoDelta
25+
import org.apache.spark.sql.delta.rapids.{GpuDeltaLog, GpuWriteIntoDelta}
26+
import org.apache.spark.sql.execution.command.RunnableCommand
27+
28+
case class GpuDeltaDynamicPartitionOverwriteCommand(
29+
gpuDeltaLog: GpuDeltaLog,
30+
table: NamedRelation,
31+
deltaTable: DeltaTableV2,
32+
query: LogicalPlan,
33+
writeOptions: Map[String, String],
34+
isByName: Boolean,
35+
analyzedQuery: Option[LogicalPlan] = None) extends RunnableCommand with V2WriteCommand {
36+
37+
override def child: LogicalPlan = query
38+
39+
override def withNewQuery(newQuery: LogicalPlan): V2WriteCommand = {
40+
copy(query = newQuery)
41+
}
42+
43+
override def withNewTable(newTable: NamedRelation): V2WriteCommand = {
44+
copy(table = newTable)
45+
}
46+
47+
override def storeAnalyzedQuery(): Command = copy(analyzedQuery = Some(query))
48+
49+
override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = {
50+
copy(query = newChild)
51+
}
52+
53+
override def run(sparkSession: SparkSession): Seq[Row] = {
54+
val deltaOptions = new DeltaOptions(
55+
CaseInsensitiveMap[String](
56+
deltaTable.options ++
57+
writeOptions ++
58+
Seq(DeltaOptions.PARTITION_OVERWRITE_MODE_OPTION ->
59+
DeltaOptions.PARTITION_OVERWRITE_MODE_DYNAMIC)),
60+
sparkSession.sessionState.conf)
61+
62+
GpuWriteIntoDelta(
63+
gpuDeltaLog,
64+
WriteIntoDelta(
65+
gpuDeltaLog.deltaLog,
66+
SaveMode.Overwrite,
67+
deltaOptions,
68+
partitionColumns = Nil,
69+
deltaTable.deltaLog.unsafeVolatileSnapshot.metadata.configuration,
70+
Dataset.ofRows(sparkSession, query),
71+
deltaTable.catalogTable
72+
)
73+
).run(sparkSession)
74+
}
75+
}

‎docs/additional-functionality/shuffle-docker-examples/Dockerfile.rocky_no_rdma‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,17 +15,19 @@
1515

1616
# Sample Dockerfile to install UCX in a Rocky Linux 8 image.
1717
#
18-
# The parameters are:
19-
# - CUDA_VER: 13.0.1 by default
20-
# - UCX_VER, UCX_CUDA_VER, and UCX_ARCH:
18+
# The parameters are:
19+
# - CUDA_VER: 13.0.1 by default - CUDA runtime version installed in the container
20+
# - UCX_VER, UCX_CUDA_VER, and UCX_ARCH:
2121
# Used to pick a package matching a specific UCX version and
22-
# CUDA runtime from the UCX github repo.
22+
# CUDA runtime from the UCX github repo. UCX_CUDA_VER is the CUDA version
23+
# that UCX is built for (can be older than CUDA_VER due to backwards compatibility;
24+
# CUDA 13 UCX builds are not available for Rocky Linux, hence UCX_CUDA_VER=11).
2325
# See: https://github.com/openucx/ucx/releases/
2426
# - ROCKY_VER: Rocky Linux OS version
2527

2628
ARG CUDA_VER=13.0.1
2729
ARG UCX_VER=1.19.1-rc1
28-
ARG UCX_CUDA_VER=13
30+
ARG UCX_CUDA_VER=11
2931
ARG UCX_ARCH=x86_64
3032
ARG ROCKY_VER=8
3133
FROM nvidia/cuda:${CUDA_VER}-runtime-rockylinux${ROCKY_VER}

‎docs/additional-functionality/shuffle-docker-examples/Dockerfile.rocky_rdma‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,17 +15,19 @@
1515

1616
# Sample Dockerfile to install UCX in a Rocky Linux 8 image with RDMA support.
1717
#
18-
# The parameters are:
19-
# - CUDA_VER: 13.0.1 by default
20-
# - UCX_VER, UCX_CUDA_VER, and UCX_ARCH:
18+
# The parameters are:
19+
# - CUDA_VER: 13.0.1 by default - CUDA runtime version installed in the container
20+
# - UCX_VER, UCX_CUDA_VER, and UCX_ARCH:
2121
# Used to pick a package matching a specific UCX version and
22-
# CUDA runtime from the UCX github repo.
22+
# CUDA runtime from the UCX github repo. UCX_CUDA_VER is the CUDA version
23+
# that UCX is built for (can be older than CUDA_VER due to backwards compatibility;
24+
# CUDA 13 UCX builds are not available for Rocky Linux, hence UCX_CUDA_VER=11).
2325
# See: https://github.com/openucx/ucx/releases/
2426
# - ROCKY_VER: Rocky Linux OS version
2527

2628
ARG CUDA_VER=13.0.1
2729
ARG UCX_VER=1.19.1-rc1
28-
ARG UCX_CUDA_VER=13
30+
ARG UCX_CUDA_VER=11
2931
ARG UCX_ARCH=x86_64
3032
ARG ROCKY_VER=8
3133
FROM nvidia/cuda:${CUDA_VER}-runtime-rockylinux${ROCKY_VER}

‎iceberg/src/main/scala/com/nvidia/spark/rapids/iceberg/IcebergProviderImpl.scala‎

Lines changed: 39 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,17 @@ package com.nvidia.spark.rapids.iceberg
1919
import scala.reflect.ClassTag
2020
import scala.util.{Failure, Success, Try}
2121

22-
import com.nvidia.spark.rapids.{AppendDataExecMeta, FileFormatChecks, GpuExec, GpuExpression, GpuRowToColumnarExec, GpuScan, IcebergFormatType, OverwritePartitionsDynamicExecMeta, RapidsConf, ReadFileOp, ScanMeta, ScanRule, ShimReflectionUtils, StaticInvokeMeta, TargetSize, WriteFileOp}
22+
import com.nvidia.spark.rapids.{AppendDataExecMeta, AtomicCreateTableAsSelectExecMeta, FileFormatChecks, GpuExec, GpuExpression, GpuRowToColumnarExec, GpuScan, IcebergFormatType, OverwritePartitionsDynamicExecMeta, RapidsConf, ReadFileOp, ScanMeta, ScanRule, ShimReflectionUtils, StaticInvokeMeta, TargetSize, WriteFileOp}
2323
import org.apache.iceberg.spark.functions.{BucketFunction, GpuBucketExpression}
2424
import org.apache.iceberg.spark.source.{GpuSparkBatchQueryScan, GpuSparkWrite}
25+
import org.apache.iceberg.spark.supportsCatalog
2526

2627
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
2728
import org.apache.spark.sql.connector.read.Scan
2829
import org.apache.spark.sql.connector.write.Write
2930
import org.apache.spark.sql.execution.SparkPlan
30-
import org.apache.spark.sql.execution.datasources.v2.{AppendDataExec, GpuAppendDataExec, GpuOverwritePartitionsDynamicExec, OverwritePartitionsDynamicExec}
31+
import org.apache.spark.sql.execution.datasources.v2.{AppendDataExec, AtomicCreateTableAsSelectExec, GpuAppendDataExec, GpuOverwritePartitionsDynamicExec, OverwritePartitionsDynamicExec}
32+
import org.apache.spark.sql.execution.datasources.v2.rapids.GpuAtomicCreateTableAsSelectExec
3133

3234
class IcebergProviderImpl extends IcebergProvider {
3335
override def getScans: Map[Class[_ <: Scan], ScanRule[_ <: Scan]] = {
@@ -98,6 +100,41 @@ class IcebergProviderImpl extends IcebergProvider {
98100
GpuSparkWrite.supports(write)
99101
}
100102

103+
override def isSupportedCatalog(catalogClass: Class[_]): Boolean = {
104+
supportsCatalog(catalogClass)
105+
}
106+
107+
override def tagForGpu(
108+
cpuExec: AtomicCreateTableAsSelectExec,
109+
meta: AtomicCreateTableAsSelectExecMeta): Unit = {
110+
if (!meta.conf.isIcebergEnabled) {
111+
meta.willNotWorkOnGpu("Iceberg input and output has been disabled. To enable set " +
112+
s"${RapidsConf.ENABLE_ICEBERG.key} to true")
113+
}
114+
115+
if (!meta.conf.isIcebergWriteEnabled) {
116+
meta.willNotWorkOnGpu("Iceberg output has been disabled. To enable set " +
117+
s"${RapidsConf.ENABLE_ICEBERG_WRITE.key} to true")
118+
}
119+
120+
FileFormatChecks.tag(meta, cpuExec.query.schema, IcebergFormatType, WriteFileOp)
121+
122+
GpuSparkWrite.tagForGpuCtas(cpuExec, meta)
123+
}
124+
125+
override def convertToGpu(
126+
cpuExec: AtomicCreateTableAsSelectExec,
127+
meta: AtomicCreateTableAsSelectExecMeta): GpuExec = {
128+
GpuAtomicCreateTableAsSelectExec(
129+
cpuExec.catalog,
130+
cpuExec.ident,
131+
cpuExec.partitioning,
132+
cpuExec.query,
133+
cpuExec.tableSpec,
134+
cpuExec.writeOptions,
135+
cpuExec.ifNotExists)
136+
}
137+
101138
override def tagForGpu(cpuExec: AppendDataExec, meta: AppendDataExecMeta): Unit = {
102139
if (!meta.conf.isIcebergEnabled) {
103140
meta.willNotWorkOnGpu("Iceberg input and output has been disabled. To enable set " +

‎iceberg/src/main/scala/org/apache/iceberg/spark/package.scala‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
/*
2+
* Copyright (c) 2025, NVIDIA CORPORATION.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.apache.iceberg
18+
19+
package object spark {
20+
def supportsCatalog(catalogClass: Class[_]): Boolean = {
21+
classOf[BaseCatalog].isAssignableFrom(catalogClass)
22+
}
23+
}

0 commit comments

Comments
 (0)