This is an automated email from the ASF dual-hosted git repository.
jackylee-ch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new ca82b7721f [GLUTEN-12698][CORE] Add Iceberg read/write offload
switches (#12699)
ca82b7721f is described below
commit ca82b7721f93e068caab8c96efbbdf2a9499bdb0
Author: jackylee <[email protected]>
AuthorDate: Tue Aug 11 09:31:48 2026 +0800
[GLUTEN-12698][CORE] Add Iceberg read/write offload switches (#12699)
---
.../gluten/extension/OffloadIcebergWrite.scala | 67 +++++++++++++++-------
.../execution/enhanced/VeloxIcebergSuite.scala | 31 ++++++++++
docs/get-started/VeloxIceberg.md | 8 +++
.../apache/gluten/config/GlutenIcebergConfig.scala | 49 ++++++++++++++++
.../gluten/extension/OffloadIcebergScan.scala | 15 +++--
.../org/apache/gluten/execution/IcebergSuite.scala | 51 ++++++++++++++++
6 files changed, 195 insertions(+), 26 deletions(-)
diff --git
a/backends-velox/src-iceberg/main/scala/org/apache/gluten/extension/OffloadIcebergWrite.scala
b/backends-velox/src-iceberg/main/scala/org/apache/gluten/extension/OffloadIcebergWrite.scala
index e0fcbae6c6..05d19334d2 100644
---
a/backends-velox/src-iceberg/main/scala/org/apache/gluten/extension/OffloadIcebergWrite.scala
+++
b/backends-velox/src-iceberg/main/scala/org/apache/gluten/extension/OffloadIcebergWrite.scala
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.extension
-import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.config.{GlutenConfig, GlutenIcebergConfig}
import org.apache.gluten.execution._
import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
@@ -29,42 +29,67 @@ import org.apache.spark.sql.execution.datasources.v2._
import org.apache.iceberg.spark.source.IcebergWriteUtil.supportsWrite
case class OffloadIcebergAppend() extends OffloadSingleNode {
- override def offload(plan: SparkPlan): SparkPlan = plan match {
- case a: AppendDataExec if supportsWrite(a.write) =>
- VeloxIcebergAppendDataExec(a)
- case other => other
+ override def offload(plan: SparkPlan): SparkPlan = {
+ if (!GlutenIcebergConfig.get.enableNativeWrite) {
+ return plan
+ }
+ plan match {
+ case a: AppendDataExec if supportsWrite(a.write) =>
+ VeloxIcebergAppendDataExec(a)
+ case other => other
+ }
}
}
case class OffloadIcebergReplaceData() extends OffloadSingleNode {
- override def offload(plan: SparkPlan): SparkPlan = plan match {
- case r: ReplaceDataExec if supportsWrite(r.write) =>
- VeloxIcebergReplaceDataExec(r)
- case other => other
+ override def offload(plan: SparkPlan): SparkPlan = {
+ if (!GlutenIcebergConfig.get.enableNativeWrite) {
+ return plan
+ }
+ plan match {
+ case r: ReplaceDataExec if supportsWrite(r.write) =>
+ VeloxIcebergReplaceDataExec(r)
+ case other => other
+ }
}
}
case class OffloadIcebergOverwrite() extends OffloadSingleNode {
- override def offload(plan: SparkPlan): SparkPlan = plan match {
- case r: OverwriteByExpressionExec if supportsWrite(r.write) =>
- VeloxIcebergOverwriteByExpressionExec(r)
- case other => other
+ override def offload(plan: SparkPlan): SparkPlan = {
+ if (!GlutenIcebergConfig.get.enableNativeWrite) {
+ return plan
+ }
+ plan match {
+ case r: OverwriteByExpressionExec if supportsWrite(r.write) =>
+ VeloxIcebergOverwriteByExpressionExec(r)
+ case other => other
+ }
}
}
case class OffloadIcebergOverwritePartitionsDynamic() extends
OffloadSingleNode {
- override def offload(plan: SparkPlan): SparkPlan = plan match {
- case r: OverwritePartitionsDynamicExec if supportsWrite(r.write) =>
- VeloxIcebergOverwritePartitionsDynamicExec(r)
- case other => other
+ override def offload(plan: SparkPlan): SparkPlan = {
+ if (!GlutenIcebergConfig.get.enableNativeWrite) {
+ return plan
+ }
+ plan match {
+ case r: OverwritePartitionsDynamicExec if supportsWrite(r.write) =>
+ VeloxIcebergOverwritePartitionsDynamicExec(r)
+ case other => other
+ }
}
}
case class OffloadIcebergWriteToDataSourceV2() extends OffloadSingleNode {
- override def offload(plan: SparkPlan): SparkPlan = plan match {
- case r: WriteToDataSourceV2Exec =>
- VeloxIcebergWriteToDataSourceV2Exec(r).getOrElse(r)
- case other => other
+ override def offload(plan: SparkPlan): SparkPlan = {
+ if (!GlutenIcebergConfig.get.enableNativeWrite) {
+ return plan
+ }
+ plan match {
+ case r: WriteToDataSourceV2Exec =>
+ VeloxIcebergWriteToDataSourceV2Exec(r).getOrElse(r)
+ case other => other
+ }
}
}
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
index 3a8c48d5e9..f3d8e290d4 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
+++
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
@@ -16,6 +16,7 @@
*/
package org.apache.gluten.execution.enhanced
+import org.apache.gluten.config.GlutenIcebergConfig
import org.apache.gluten.config.VeloxConfig.MAX_TARGET_FILE_SIZE_SESSION
import org.apache.gluten.execution._
import org.apache.gluten.tags.EnhancedFeaturesTest
@@ -775,4 +776,34 @@ class VeloxIcebergSuite extends IcebergSuite {
}
}
}
+
+ test("iceberg write falls back when native write is disabled") {
+ withTable("iceberg_write_switch_tbl") {
+ spark.sql("CREATE TABLE iceberg_write_switch_tbl (a INT, b STRING) USING
iceberg")
+
+ withSQLConf(GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key -> "false") {
+ val df = spark.sql("INSERT INTO iceberg_write_switch_tbl VALUES (1,
'hello')")
+ val commandPlan =
+
df.queryExecution.executedPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan
+ assert(
+ !commandPlan.isInstanceOf[VeloxIcebergAppendDataExec],
+ s"Iceberg write should not be offloaded when native write is
disabled: $commandPlan")
+ assert(commandPlan.isInstanceOf[AppendDataExec])
+ }
+
+ // Reads stay offloaded: the write switch must not affect the read path.
+ runQueryAndCompare("SELECT * FROM iceberg_write_switch_tbl") {
+ checkGlutenPlan[IcebergScanTransformer]
+ }
+
+ // The switch is dynamic: offload resumes once it is back to the default.
+ TestUtils.checkExecutedPlanContains[VeloxIcebergAppendDataExec](
+ spark,
+ "INSERT INTO iceberg_write_switch_tbl VALUES (2, 'world')")
+
+ checkAnswer(
+ spark.sql("SELECT * FROM iceberg_write_switch_tbl ORDER BY a"),
+ Seq(Row(1, "hello"), Row(2, "world")))
+ }
+ }
}
diff --git a/docs/get-started/VeloxIceberg.md b/docs/get-started/VeloxIceberg.md
index aea2e89c76..42901a26cf 100644
--- a/docs/get-started/VeloxIceberg.md
+++ b/docs/get-started/VeloxIceberg.md
@@ -98,6 +98,14 @@ Gluten uses column name to match the parquet file, so if the
column is renamed o
the added column name is same to the deleted column, the scan will fall back.
## Configuration
+### Gluten Options
+| Gluten option | Default | Description |
+| --- | --- | --- |
+| spark.gluten.sql.columnar.iceberg.enableNativeRead | true | Enable
offloading Iceberg scans to the native backend. When disabled, Iceberg scans
fall back to vanilla Spark while scans of other formats stay offloaded. |
+| spark.gluten.sql.columnar.iceberg.enableNativeWrite | true | Enable
offloading Iceberg writes to the native backend. When disabled, Iceberg writes
fall back to vanilla Spark. Note the Velox backend additionally requires
`spark.gluten.sql.enable.enhancedFeatures` to be enabled. |
+
+Both options are runtime modifiable, so they can be flipped per session with
`SET`.
+
### Catalogs
All the catalog configurations are transparent to Gluten
diff --git
a/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala
b/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala
new file mode 100644
index 0000000000..c148ff135b
--- /dev/null
+++
b/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.gluten.config
+
+import org.apache.spark.sql.internal.SQLConf
+
+class GlutenIcebergConfig(conf: SQLConf) extends GlutenCoreConfig(conf) {
+ import GlutenIcebergConfig._
+
+ def enableNativeRead: Boolean = getConf(ENABLE_NATIVE_READ)
+
+ def enableNativeWrite: Boolean = getConf(ENABLE_NATIVE_WRITE)
+}
+
+object GlutenIcebergConfig extends ConfigRegistry {
+
+ def get: GlutenIcebergConfig = {
+ new GlutenIcebergConfig(SQLConf.get)
+ }
+
+ val ENABLE_NATIVE_READ: ConfigEntry[Boolean] =
+ buildConf("spark.gluten.sql.columnar.iceberg.enableNativeRead")
+ .doc("Enable offloading Iceberg scans to the native backend. When
disabled, Iceberg scans" +
+ " fall back to vanilla Spark while scans of other formats stay
offloaded.")
+ .booleanConf
+ .createWithDefault(true)
+
+ val ENABLE_NATIVE_WRITE: ConfigEntry[Boolean] =
+ buildConf("spark.gluten.sql.columnar.iceberg.enableNativeWrite")
+ .doc("Enable offloading Iceberg writes to the native backend. When
disabled, Iceberg" +
+ " writes fall back to vanilla Spark. Note the Velox backend
additionally requires" +
+ " spark.gluten.sql.enable.enhancedFeatures to be enabled.")
+ .booleanConf
+ .createWithDefault(true)
+}
diff --git
a/gluten-iceberg/src/main/scala/org/apache/gluten/extension/OffloadIcebergScan.scala
b/gluten-iceberg/src/main/scala/org/apache/gluten/extension/OffloadIcebergScan.scala
index 505e063146..c043449dc9 100644
---
a/gluten-iceberg/src/main/scala/org/apache/gluten/extension/OffloadIcebergScan.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/gluten/extension/OffloadIcebergScan.scala
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.extension
-import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.config.{GlutenConfig, GlutenIcebergConfig}
import org.apache.gluten.execution.IcebergScanTransformer
import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
@@ -27,10 +27,15 @@ import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
case class OffloadIcebergScan() extends OffloadSingleNode {
- override def offload(plan: SparkPlan): SparkPlan = plan match {
- case scan: BatchScanExec if
IcebergScanTransformer.supportsBatchScan(scan.scan) =>
- IcebergScanTransformer(scan)
- case other => other
+ override def offload(plan: SparkPlan): SparkPlan = {
+ if (!GlutenIcebergConfig.get.enableNativeRead) {
+ return plan
+ }
+ plan match {
+ case scan: BatchScanExec if
IcebergScanTransformer.supportsBatchScan(scan.scan) =>
+ IcebergScanTransformer(scan)
+ case other => other
+ }
}
}
diff --git
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
index 7f9b0c533e..56f3fbdace 100644
---
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
+++
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
@@ -16,8 +16,11 @@
*/
package org.apache.gluten.execution
+import org.apache.gluten.config.GlutenIcebergConfig
+
import org.apache.spark.SparkConf
import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
abstract class IcebergSuite extends WholeStageTransformerSuite {
protected val rootPath: String = getClass.getResource("/").getPath
@@ -717,4 +720,52 @@ abstract class IcebergSuite extends
WholeStageTransformerSuite {
e.getCause != null && e.getCause.getMessage.contains("null"))
}
}
+
+ test("iceberg scan falls back when native read is disabled") {
+ withTable("iceberg_read_switch_tb") {
+ spark.sql("""
+ |create table iceberg_read_switch_tb using iceberg as
+ |(select 1 as col1, 2 as col2)
+ |""".stripMargin)
+
+ withSQLConf(GlutenIcebergConfig.ENABLE_NATIVE_READ.key -> "false") {
+ val df = spark.sql("select * from iceberg_read_switch_tb")
+ checkSparkPlan[BatchScanExec](df)
+ assert(
+ !getExecutedPlan(df).exists(_.isInstanceOf[IcebergScanTransformer]),
+ "Iceberg scan should not be offloaded when native read is disabled")
+ checkAnswer(df, Seq(Row(1, 2)))
+ }
+
+ // The switch is dynamic: offload resumes once it is back to the default.
+ runQueryAndCompare("select * from iceberg_read_switch_tb") {
+ checkGlutenPlan[IcebergScanTransformer]
+ }
+ }
+ }
+
+ test("disabling iceberg native read keeps other scans offloaded") {
+ withTable("iceberg_read_switch_tb") {
+ spark.sql("""
+ |create table iceberg_read_switch_tb using iceberg as
+ |(select 1 as col1)
+ |""".stripMargin)
+
+ withTempPath {
+ path =>
+ spark.range(5).toDF("col1").write.parquet(path.getCanonicalPath)
+
+ withSQLConf(GlutenIcebergConfig.ENABLE_NATIVE_READ.key -> "false") {
+ val icebergDf = spark.sql("select * from iceberg_read_switch_tb")
+ assert(
+
!getExecutedPlan(icebergDf).exists(_.isInstanceOf[IcebergScanTransformer]),
+ "Iceberg scan should fall back")
+
+ val parquetDf = spark.read.parquet(path.getCanonicalPath)
+ checkGlutenPlan[FileSourceScanExecTransformer](parquetDf)
+ assert(parquetDf.count() == 5)
+ }
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
