[AURON #2425] Support Iceberg STARTS_WITH pruning in native scan

thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala

@@ -27,7 +27,7 @@ import org.apache.iceberg.expressions.{And => IcebergAnd, BoundPredicate, Expres
 import org.apache.iceberg.spark.source.AuronIcebergSourceUtil
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.auron.{NativeConverters, Shims}
-import org.apache.spark.sql.catalyst.expressions.{And => SparkAnd, AttributeReference, EqualTo, Expression => SparkExpression, GreaterThan, GreaterThanOrEqual, In, IsNaN, IsNotNull, IsNull, LessThan, LessThanOrEqual, Literal, Not => SparkNot, Or => SparkOr}
+import org.apache.spark.sql.catalyst.expressions.{And => SparkAnd, AttributeReference, EqualTo, Expression => SparkExpression, GreaterThan, GreaterThanOrEqual, In, IsNaN, IsNotNull, IsNull, LessThan, LessThanOrEqual, Literal, Not => SparkNot, Or => SparkOr, StartsWith}
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.connector.read.{InputPartition, Scan}
 import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
@@ -752,6 +752,15 @@ object IcebergScanSupport extends Logging {
       dataType: DataType,
       op: org.apache.iceberg.expressions.Expression.Operation,
       literalValue: Any): Option[SparkExpression] = {
+    if (op == org.apache.iceberg.expressions.Expression.Operation.STARTS_WITH) {
+      if (dataType == StringType) {
+        return toLiteral(literalValue, StringType)
+          .filter(_.value != null)
+          .map(StartsWith(attr, _))
+      }
+      return None
+    }
+
     if (!supportsScanPruningLiteralType(dataType)) {
       return None
     }

thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala

@@ -466,6 +466,31 @@ class AuronIcebergIntegrationSuite
     }
   }
 
+  test("iceberg scan pushes STARTS_WITH filters into native scan pruning predicates") {
+    withTable("local.db.t_residual_starts_with") {
+      sql("create table local.db.t_residual_starts_with (id int, v string) using iceberg")
+      sql("""
+          |insert into local.db.t_residual_starts_with
+          |values (1, 'alpha'), (2, 'beta'), (3, 'atom'), (4, null)
+          |""".stripMargin)
+      val df = sql("""
+          |select * from local.db.t_residual_starts_with
+          |where v like 'a%'
+          |""".stripMargin)
+      checkAnswer(df, Seq(Row(1, "alpha"), Row(3, "atom")))
+      val nativeScanPlan = icebergScanPlan(df)
+      assert(nativeScanPlan.nonEmpty)
+      val pruningPredicateText = nativeScanPlan.get.pruningPredicates.mkString("\n")
+      assert(
+        pruningPredicateText.contains("name: \"starts_with\"") &&
+          pruningPredicateText.contains("fun: StartsWith"),
+        pruningPredicateText)
+      val plan = df.queryExecution.executedPlan.toString()
+      assert(plan.contains("NativeIcebergTableScan"))
+      assert(plan.contains("NativeFilter"))
+    }
+  }
+
   test("iceberg scan keeps native post-scan filter when only part of the predicate is pushed") {
     withTable("local.db.t_residual_partial_pushdown") {
       sql("create table local.db.t_residual_partial_pushdown (id int, v string) using iceberg")