[GLUTEN-12698][CORE] Add Iceberg read/write offload switches

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
+    }
   }
 }
 

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")))
+    }
+  }
 }

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
 

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)
+}

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
+    }
   }
 }
 

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)
+          }
+      }
+    }
+  }
 }