From 3bbb123e152330a441e81ecd3abde55f5c653878 Mon Sep 17 00:00:00 2001 From: jackylee Date: Wed, 5 Aug 2026 16:23:22 +0800 Subject: [PATCH 1/3] [GLUTEN-12698][CORE] Add Iceberg read/write offload switches Gluten had no Iceberg-specific configuration entry. Disabling Iceberg offload meant reaching for a much broader switch: `spark.gluten.sql.columnar.batchscan` for reads, which also falls back Paimon, Hudi and every other batch scan, and `spark.gluten.sql.enable.enhancedFeatures` for writes, a bundled flag documented as covering "iceberg native write and other features". Add two backend-agnostic switches, both defaulting to true so behaviour is unchanged: spark.gluten.sql.columnar.iceberg.enableNativeRead spark.gluten.sql.columnar.iceberg.enableNativeWrite They live in `gluten-iceberg` rather than a backend module because `OffloadIcebergScan` is shared by the Velox and ClickHouse components, and a backend that later gains Iceberg write support should honour the same write key instead of adding its own. Both are checked inside the offload rules rather than at rule-injection time, so they stay modifiable per session. The write switch is AND-ed with the existing `enhancedFeatures` gate rather than replacing it. Fixes #12698 --- .../extension/OffloadIcebergWrite.scala | 37 +++++++--- .../enhanced/VeloxIcebergSuite.scala | 31 ++++++++ .../extension/OffloadIcebergWriteSuite.scala | 63 ++++++++++++++++ docs/get-started/VeloxIceberg.md | 8 ++ .../gluten/config/GlutenIcebergConfig.scala | 55 ++++++++++++++ .../gluten/extension/OffloadIcebergScan.scala | 21 +++++- .../config/GlutenIcebergConfigSuite.scala | 65 +++++++++++++++++ .../gluten/execution/IcebergSuite.scala | 53 ++++++++++++++ .../extension/OffloadIcebergScanSuite.scala | 73 +++++++++++++++++++ 9 files changed, 392 insertions(+), 14 deletions(-) create mode 100644 backends-velox/src-iceberg/test/scala/org/apache/gluten/extension/OffloadIcebergWriteSuite.scala create mode 100644 gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala create mode 100644 gluten-iceberg/src/test/scala/org/apache/gluten/config/GlutenIcebergConfigSuite.scala create mode 100644 gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala 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 e0fcbae6c67..2e137b8f720 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 @@ -28,40 +28,55 @@ 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 { +/** + * Base of the Iceberg write offload rules. The switch is checked here rather than at rule injection + * time so that it stays modifiable at runtime. + */ +trait OffloadIcebergWriteBase extends OffloadSingleNode { + final override def offload(plan: SparkPlan): SparkPlan = { + if (!GlutenIcebergConfig.get.enableNativeWrite) { + return plan + } + offloadWrite(plan) + } + + protected def offloadWrite(plan: SparkPlan): SparkPlan +} + +case class OffloadIcebergAppend() extends OffloadIcebergWriteBase { + override protected def offloadWrite(plan: SparkPlan): SparkPlan = 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 class OffloadIcebergReplaceData() extends OffloadIcebergWriteBase { + override protected def offloadWrite(plan: SparkPlan): SparkPlan = 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 class OffloadIcebergOverwrite() extends OffloadIcebergWriteBase { + override protected def offloadWrite(plan: SparkPlan): SparkPlan = 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 class OffloadIcebergOverwritePartitionsDynamic() extends OffloadIcebergWriteBase { + override protected def offloadWrite(plan: SparkPlan): SparkPlan = 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 class OffloadIcebergWriteToDataSourceV2() extends OffloadIcebergWriteBase { + override protected def offloadWrite(plan: SparkPlan): SparkPlan = 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 3a8c48d5e9f..f3d8e290d42 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/backends-velox/src-iceberg/test/scala/org/apache/gluten/extension/OffloadIcebergWriteSuite.scala b/backends-velox/src-iceberg/test/scala/org/apache/gluten/extension/OffloadIcebergWriteSuite.scala new file mode 100644 index 00000000000..fb55d5be82f --- /dev/null +++ b/backends-velox/src-iceberg/test/scala/org/apache/gluten/extension/OffloadIcebergWriteSuite.scala @@ -0,0 +1,63 @@ +/* + * 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.extension + +import org.apache.gluten.config.GlutenIcebergConfig + +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan} +import org.apache.spark.sql.internal.SQLConf + +import org.scalatest.funsuite.AnyFunSuite + +class OffloadIcebergWriteSuite extends AnyFunSuite { + + private case class DummyPlan() extends LeafExecNode { + override def output: Seq[Attribute] = Seq.empty + override protected def doExecute() = throw new UnsupportedOperationException() + } + + private case class OffloadedPlan() extends LeafExecNode { + override def output: Seq[Attribute] = Seq.empty + override protected def doExecute() = throw new UnsupportedOperationException() + } + + /** Offloads unconditionally, so the only thing that can stop it is the config gate. */ + private case class AlwaysOffload() extends OffloadIcebergWriteBase { + override protected def offloadWrite(plan: SparkPlan): SparkPlan = OffloadedPlan() + } + + private def withNativeWrite[T](enabled: Boolean)(body: => T): T = { + val conf = SQLConf.get + val key = GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key + conf.setConfString(key, enabled.toString) + try body + finally conf.unsetConf(key) + } + + test("write offload rules skip offloading when native write is disabled") { + withNativeWrite(enabled = false) { + assert(AlwaysOffload().offload(DummyPlan()) === DummyPlan()) + } + } + + test("write offload rules offload when native write is enabled") { + withNativeWrite(enabled = true) { + assert(AlwaysOffload().offload(DummyPlan()) === OffloadedPlan()) + } + } +} diff --git a/docs/get-started/VeloxIceberg.md b/docs/get-started/VeloxIceberg.md index aea2e89c76f..42901a26cff 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 00000000000..101df2530eb --- /dev/null +++ b/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala @@ -0,0 +1,55 @@ +/* + * 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) +} + +/** + * Configurations of the Iceberg component. They are backend-agnostic on purpose: the read path + * ([[org.apache.gluten.extension.OffloadIcebergScan]]) is shared by all backends, and a backend + * newly supporting Iceberg write is expected to honor the same write switch rather than introduce + * its own key. + */ +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 505e0631464..a13e5da2e37 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 @@ -26,8 +26,23 @@ import org.apache.gluten.extension.injector.Injector 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 { +/** + * Base of the Iceberg scan offload rule. The switch is checked here rather than at rule injection + * time so that it stays modifiable at runtime. + */ +trait OffloadIcebergScanBase extends OffloadSingleNode { + final override def offload(plan: SparkPlan): SparkPlan = { + if (!GlutenIcebergConfig.get.enableNativeRead) { + return plan + } + offloadScan(plan) + } + + protected def offloadScan(plan: SparkPlan): SparkPlan +} + +case class OffloadIcebergScan() extends OffloadIcebergScanBase { + override protected def offloadScan(plan: SparkPlan): SparkPlan = 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/config/GlutenIcebergConfigSuite.scala b/gluten-iceberg/src/test/scala/org/apache/gluten/config/GlutenIcebergConfigSuite.scala new file mode 100644 index 00000000000..b645a88899e --- /dev/null +++ b/gluten-iceberg/src/test/scala/org/apache/gluten/config/GlutenIcebergConfigSuite.scala @@ -0,0 +1,65 @@ +/* + * 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 + +import org.scalatest.funsuite.AnyFunSuite + +class GlutenIcebergConfigSuite extends AnyFunSuite { + + test("Iceberg offload switches are backend-agnostic and enabled by default") { + assert( + GlutenIcebergConfig.ENABLE_NATIVE_READ.key === + "spark.gluten.sql.columnar.iceberg.enableNativeRead") + assert( + GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key === + "spark.gluten.sql.columnar.iceberg.enableNativeWrite") + + assert(GlutenIcebergConfig.ENABLE_NATIVE_READ.defaultValue === Some(true)) + assert(GlutenIcebergConfig.ENABLE_NATIVE_WRITE.defaultValue === Some(true)) + + assert(GlutenIcebergConfig.get.enableNativeRead) + assert(GlutenIcebergConfig.get.enableNativeWrite) + } + + test("Iceberg offload switches are read from the active SQLConf") { + val conf = SQLConf.get + Seq( + GlutenIcebergConfig.ENABLE_NATIVE_READ.key, + GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key).foreach { + key => + // Registered to SQLConf as a dynamic conf, so operators can flip it per session. + assert(conf.isModifiable(key), s"$key should be runtime modifiable") + } + + try { + conf.setConfString(GlutenIcebergConfig.ENABLE_NATIVE_READ.key, "false") + assert(!GlutenIcebergConfig.get.enableNativeRead) + assert(GlutenIcebergConfig.get.enableNativeWrite) + + conf.setConfString(GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key, "false") + assert(!GlutenIcebergConfig.get.enableNativeWrite) + } finally { + conf.unsetConf(GlutenIcebergConfig.ENABLE_NATIVE_READ.key) + conf.unsetConf(GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key) + } + + assert(GlutenIcebergConfig.get.enableNativeRead) + assert(GlutenIcebergConfig.get.enableNativeWrite) + } +} 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 7f9b0c533e0..499aef5f9a2 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,54 @@ 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") + + // The switch is scoped to Iceberg: scans of other formats keep being offloaded, which + // is the whole point of not reusing spark.gluten.sql.columnar.batchscan for this. + val parquetDf = spark.read.parquet(path.getCanonicalPath) + checkGlutenPlan[FileSourceScanExecTransformer](parquetDf) + assert(parquetDf.count() == 5) + } + } + } + } } diff --git a/gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala b/gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala new file mode 100644 index 00000000000..d7844c40efc --- /dev/null +++ b/gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala @@ -0,0 +1,73 @@ +/* + * 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.extension + +import org.apache.gluten.config.GlutenIcebergConfig + +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan} +import org.apache.spark.sql.internal.SQLConf + +import org.scalatest.funsuite.AnyFunSuite + +class OffloadIcebergScanSuite extends AnyFunSuite { + + private case class DummyPlan() extends LeafExecNode { + override def output: Seq[Attribute] = Seq.empty + override protected def doExecute() = throw new UnsupportedOperationException() + } + + private case class OffloadedPlan() extends LeafExecNode { + override def output: Seq[Attribute] = Seq.empty + override protected def doExecute() = throw new UnsupportedOperationException() + } + + /** Offloads unconditionally, so the only thing that can stop it is the config gate. */ + private case class AlwaysOffload() extends OffloadIcebergScanBase { + override protected def offloadScan(plan: SparkPlan): SparkPlan = OffloadedPlan() + } + + private def withNativeRead[T](enabled: Boolean)(body: => T): T = { + val conf = SQLConf.get + val key = GlutenIcebergConfig.ENABLE_NATIVE_READ.key + conf.setConfString(key, enabled.toString) + try body + finally conf.unsetConf(key) + } + + test("scan offload rule skips offloading when native read is disabled") { + withNativeRead(enabled = false) { + assert(AlwaysOffload().offload(DummyPlan()) === DummyPlan()) + } + } + + test("scan offload rule offloads when native read is enabled") { + withNativeRead(enabled = true) { + assert(AlwaysOffload().offload(DummyPlan()) === OffloadedPlan()) + } + } + + test("native read switch is consulted per call, not cached at rule construction") { + val rule = AlwaysOffload() + withNativeRead(enabled = false) { + assert(rule.offload(DummyPlan()) === DummyPlan()) + } + withNativeRead(enabled = true) { + assert(rule.offload(DummyPlan()) === OffloadedPlan()) + } + } +} From b0a4a27759a5a6a06654cc2fb1e2ce630aa9e0fe Mon Sep 17 00:00:00 2001 From: jackylee-ch Date: Thu, 6 Aug 2026 02:02:04 +0800 Subject: [PATCH 2/3] [GLUTEN-12698][CORE] Address review comments Drop the `OffloadIcebergScanBase` / `OffloadIcebergWriteBase` traits and inline the config check into each offload rule, following the existing `OffloadDeltaCommand` precedent. The traits existed only as an injection point for the unit suites. Remove `OffloadIcebergScanSuite`, `OffloadIcebergWriteSuite` and `GlutenIcebergConfigSuite`. The switches are already covered end to end by the integration tests added to `IcebergSuite` and `VeloxIcebergSuite` in this PR. --- .../extension/OffloadIcebergWrite.scala | 87 +++++++++++-------- .../extension/OffloadIcebergWriteSuite.scala | 63 -------------- .../gluten/extension/OffloadIcebergScan.scala | 26 ++---- .../config/GlutenIcebergConfigSuite.scala | 65 -------------- .../extension/OffloadIcebergScanSuite.scala | 73 ---------------- 5 files changed, 59 insertions(+), 255 deletions(-) delete mode 100644 backends-velox/src-iceberg/test/scala/org/apache/gluten/extension/OffloadIcebergWriteSuite.scala delete mode 100644 gluten-iceberg/src/test/scala/org/apache/gluten/config/GlutenIcebergConfigSuite.scala delete mode 100644 gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala 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 2e137b8f720..feea1e821bf 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 @@ -28,58 +28,71 @@ import org.apache.spark.sql.execution.datasources.v2._ import org.apache.iceberg.spark.source.IcebergWriteUtil.supportsWrite -/** - * Base of the Iceberg write offload rules. The switch is checked here rather than at rule injection - * time so that it stays modifiable at runtime. - */ -trait OffloadIcebergWriteBase extends OffloadSingleNode { - final override def offload(plan: SparkPlan): SparkPlan = { +// The write switch is checked inside each rule below rather than at rule injection time so that +// it stays modifiable at runtime. + +case class OffloadIcebergAppend() extends OffloadSingleNode { + override def offload(plan: SparkPlan): SparkPlan = { if (!GlutenIcebergConfig.get.enableNativeWrite) { return plan } - offloadWrite(plan) - } - - protected def offloadWrite(plan: SparkPlan): SparkPlan -} - -case class OffloadIcebergAppend() extends OffloadIcebergWriteBase { - override protected def offloadWrite(plan: SparkPlan): SparkPlan = plan match { - case a: AppendDataExec if supportsWrite(a.write) => - VeloxIcebergAppendDataExec(a) - case other => other + plan match { + case a: AppendDataExec if supportsWrite(a.write) => + VeloxIcebergAppendDataExec(a) + case other => other + } } } -case class OffloadIcebergReplaceData() extends OffloadIcebergWriteBase { - override protected def offloadWrite(plan: SparkPlan): SparkPlan = plan match { - case r: ReplaceDataExec if supportsWrite(r.write) => - VeloxIcebergReplaceDataExec(r) - case other => other +case class OffloadIcebergReplaceData() extends OffloadSingleNode { + 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 OffloadIcebergWriteBase { - override protected def offloadWrite(plan: SparkPlan): SparkPlan = plan match { - case r: OverwriteByExpressionExec if supportsWrite(r.write) => - VeloxIcebergOverwriteByExpressionExec(r) - case other => other +case class OffloadIcebergOverwrite() extends OffloadSingleNode { + 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 OffloadIcebergWriteBase { - override protected def offloadWrite(plan: SparkPlan): SparkPlan = plan match { - case r: OverwritePartitionsDynamicExec if supportsWrite(r.write) => - VeloxIcebergOverwritePartitionsDynamicExec(r) - case other => other +case class OffloadIcebergOverwritePartitionsDynamic() extends OffloadSingleNode { + 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 OffloadIcebergWriteBase { - override protected def offloadWrite(plan: SparkPlan): SparkPlan = plan match { - case r: WriteToDataSourceV2Exec => - VeloxIcebergWriteToDataSourceV2Exec(r).getOrElse(r) - case other => other +case class OffloadIcebergWriteToDataSourceV2() extends OffloadSingleNode { + 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/extension/OffloadIcebergWriteSuite.scala b/backends-velox/src-iceberg/test/scala/org/apache/gluten/extension/OffloadIcebergWriteSuite.scala deleted file mode 100644 index fb55d5be82f..00000000000 --- a/backends-velox/src-iceberg/test/scala/org/apache/gluten/extension/OffloadIcebergWriteSuite.scala +++ /dev/null @@ -1,63 +0,0 @@ -/* - * 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.extension - -import org.apache.gluten.config.GlutenIcebergConfig - -import org.apache.spark.sql.catalyst.expressions.Attribute -import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan} -import org.apache.spark.sql.internal.SQLConf - -import org.scalatest.funsuite.AnyFunSuite - -class OffloadIcebergWriteSuite extends AnyFunSuite { - - private case class DummyPlan() extends LeafExecNode { - override def output: Seq[Attribute] = Seq.empty - override protected def doExecute() = throw new UnsupportedOperationException() - } - - private case class OffloadedPlan() extends LeafExecNode { - override def output: Seq[Attribute] = Seq.empty - override protected def doExecute() = throw new UnsupportedOperationException() - } - - /** Offloads unconditionally, so the only thing that can stop it is the config gate. */ - private case class AlwaysOffload() extends OffloadIcebergWriteBase { - override protected def offloadWrite(plan: SparkPlan): SparkPlan = OffloadedPlan() - } - - private def withNativeWrite[T](enabled: Boolean)(body: => T): T = { - val conf = SQLConf.get - val key = GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key - conf.setConfString(key, enabled.toString) - try body - finally conf.unsetConf(key) - } - - test("write offload rules skip offloading when native write is disabled") { - withNativeWrite(enabled = false) { - assert(AlwaysOffload().offload(DummyPlan()) === DummyPlan()) - } - } - - test("write offload rules offload when native write is enabled") { - withNativeWrite(enabled = true) { - assert(AlwaysOffload().offload(DummyPlan()) === OffloadedPlan()) - } - } -} 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 a13e5da2e37..48291c6e737 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 @@ -26,26 +26,18 @@ import org.apache.gluten.extension.injector.Injector import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.execution.datasources.v2.BatchScanExec -/** - * Base of the Iceberg scan offload rule. The switch is checked here rather than at rule injection - * time so that it stays modifiable at runtime. - */ -trait OffloadIcebergScanBase extends OffloadSingleNode { - final override def offload(plan: SparkPlan): SparkPlan = { +case class OffloadIcebergScan() extends OffloadSingleNode { + override def offload(plan: SparkPlan): SparkPlan = { + // The switch is checked here rather than at rule injection time so that it stays + // modifiable at runtime. if (!GlutenIcebergConfig.get.enableNativeRead) { return plan } - offloadScan(plan) - } - - protected def offloadScan(plan: SparkPlan): SparkPlan -} - -case class OffloadIcebergScan() extends OffloadIcebergScanBase { - override protected def offloadScan(plan: SparkPlan): SparkPlan = plan match { - case scan: BatchScanExec if IcebergScanTransformer.supportsBatchScan(scan.scan) => - IcebergScanTransformer(scan) - case other => other + 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/config/GlutenIcebergConfigSuite.scala b/gluten-iceberg/src/test/scala/org/apache/gluten/config/GlutenIcebergConfigSuite.scala deleted file mode 100644 index b645a88899e..00000000000 --- a/gluten-iceberg/src/test/scala/org/apache/gluten/config/GlutenIcebergConfigSuite.scala +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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 - -import org.scalatest.funsuite.AnyFunSuite - -class GlutenIcebergConfigSuite extends AnyFunSuite { - - test("Iceberg offload switches are backend-agnostic and enabled by default") { - assert( - GlutenIcebergConfig.ENABLE_NATIVE_READ.key === - "spark.gluten.sql.columnar.iceberg.enableNativeRead") - assert( - GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key === - "spark.gluten.sql.columnar.iceberg.enableNativeWrite") - - assert(GlutenIcebergConfig.ENABLE_NATIVE_READ.defaultValue === Some(true)) - assert(GlutenIcebergConfig.ENABLE_NATIVE_WRITE.defaultValue === Some(true)) - - assert(GlutenIcebergConfig.get.enableNativeRead) - assert(GlutenIcebergConfig.get.enableNativeWrite) - } - - test("Iceberg offload switches are read from the active SQLConf") { - val conf = SQLConf.get - Seq( - GlutenIcebergConfig.ENABLE_NATIVE_READ.key, - GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key).foreach { - key => - // Registered to SQLConf as a dynamic conf, so operators can flip it per session. - assert(conf.isModifiable(key), s"$key should be runtime modifiable") - } - - try { - conf.setConfString(GlutenIcebergConfig.ENABLE_NATIVE_READ.key, "false") - assert(!GlutenIcebergConfig.get.enableNativeRead) - assert(GlutenIcebergConfig.get.enableNativeWrite) - - conf.setConfString(GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key, "false") - assert(!GlutenIcebergConfig.get.enableNativeWrite) - } finally { - conf.unsetConf(GlutenIcebergConfig.ENABLE_NATIVE_READ.key) - conf.unsetConf(GlutenIcebergConfig.ENABLE_NATIVE_WRITE.key) - } - - assert(GlutenIcebergConfig.get.enableNativeRead) - assert(GlutenIcebergConfig.get.enableNativeWrite) - } -} diff --git a/gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala b/gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala deleted file mode 100644 index d7844c40efc..00000000000 --- a/gluten-iceberg/src/test/scala/org/apache/gluten/extension/OffloadIcebergScanSuite.scala +++ /dev/null @@ -1,73 +0,0 @@ -/* - * 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.extension - -import org.apache.gluten.config.GlutenIcebergConfig - -import org.apache.spark.sql.catalyst.expressions.Attribute -import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan} -import org.apache.spark.sql.internal.SQLConf - -import org.scalatest.funsuite.AnyFunSuite - -class OffloadIcebergScanSuite extends AnyFunSuite { - - private case class DummyPlan() extends LeafExecNode { - override def output: Seq[Attribute] = Seq.empty - override protected def doExecute() = throw new UnsupportedOperationException() - } - - private case class OffloadedPlan() extends LeafExecNode { - override def output: Seq[Attribute] = Seq.empty - override protected def doExecute() = throw new UnsupportedOperationException() - } - - /** Offloads unconditionally, so the only thing that can stop it is the config gate. */ - private case class AlwaysOffload() extends OffloadIcebergScanBase { - override protected def offloadScan(plan: SparkPlan): SparkPlan = OffloadedPlan() - } - - private def withNativeRead[T](enabled: Boolean)(body: => T): T = { - val conf = SQLConf.get - val key = GlutenIcebergConfig.ENABLE_NATIVE_READ.key - conf.setConfString(key, enabled.toString) - try body - finally conf.unsetConf(key) - } - - test("scan offload rule skips offloading when native read is disabled") { - withNativeRead(enabled = false) { - assert(AlwaysOffload().offload(DummyPlan()) === DummyPlan()) - } - } - - test("scan offload rule offloads when native read is enabled") { - withNativeRead(enabled = true) { - assert(AlwaysOffload().offload(DummyPlan()) === OffloadedPlan()) - } - } - - test("native read switch is consulted per call, not cached at rule construction") { - val rule = AlwaysOffload() - withNativeRead(enabled = false) { - assert(rule.offload(DummyPlan()) === DummyPlan()) - } - withNativeRead(enabled = true) { - assert(rule.offload(DummyPlan()) === OffloadedPlan()) - } - } -} From 7ff8cac2fd632af7dd209c4eb830bfd2aaffcca9 Mon Sep 17 00:00:00 2001 From: jackylee Date: Thu, 6 Aug 2026 10:25:35 +0800 Subject: [PATCH 3/3] [GLUTEN-12698][CORE] Remove redundant comments per review --- .../org/apache/gluten/extension/OffloadIcebergWrite.scala | 3 --- .../org/apache/gluten/config/GlutenIcebergConfig.scala | 6 ------ .../org/apache/gluten/extension/OffloadIcebergScan.scala | 2 -- .../scala/org/apache/gluten/execution/IcebergSuite.scala | 2 -- 4 files changed, 13 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 feea1e821bf..05d19334d2d 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 @@ -28,9 +28,6 @@ import org.apache.spark.sql.execution.datasources.v2._ import org.apache.iceberg.spark.source.IcebergWriteUtil.supportsWrite -// The write switch is checked inside each rule below rather than at rule injection time so that -// it stays modifiable at runtime. - case class OffloadIcebergAppend() extends OffloadSingleNode { override def offload(plan: SparkPlan): SparkPlan = { if (!GlutenIcebergConfig.get.enableNativeWrite) { 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 index 101df2530eb..c148ff135bd 100644 --- a/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala +++ b/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala @@ -26,12 +26,6 @@ class GlutenIcebergConfig(conf: SQLConf) extends GlutenCoreConfig(conf) { def enableNativeWrite: Boolean = getConf(ENABLE_NATIVE_WRITE) } -/** - * Configurations of the Iceberg component. They are backend-agnostic on purpose: the read path - * ([[org.apache.gluten.extension.OffloadIcebergScan]]) is shared by all backends, and a backend - * newly supporting Iceberg write is expected to honor the same write switch rather than introduce - * its own key. - */ object GlutenIcebergConfig extends ConfigRegistry { def get: GlutenIcebergConfig = { 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 48291c6e737..c043449dc95 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 @@ -28,8 +28,6 @@ import org.apache.spark.sql.execution.datasources.v2.BatchScanExec case class OffloadIcebergScan() extends OffloadSingleNode { override def offload(plan: SparkPlan): SparkPlan = { - // The switch is checked here rather than at rule injection time so that it stays - // modifiable at runtime. if (!GlutenIcebergConfig.get.enableNativeRead) { return plan } 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 499aef5f9a2..56f3fbdace7 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 @@ -761,8 +761,6 @@ abstract class IcebergSuite extends WholeStageTransformerSuite { !getExecutedPlan(icebergDf).exists(_.isInstanceOf[IcebergScanTransformer]), "Iceberg scan should fall back") - // The switch is scoped to Iceberg: scans of other formats keep being offloaded, which - // is the whole point of not reusing spark.gluten.sql.columnar.batchscan for this. val parquetDf = spark.read.parquet(path.getCanonicalPath) checkGlutenPlan[FileSourceScanExecTransformer](parquetDf) assert(parquetDf.count() == 5)