diff --git a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala b/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala index 31d877930..fb23ed43d 100644 --- a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala +++ b/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 } diff --git a/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala b/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala index 6d342bd48..c115331a2 100644 --- a/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala +++ b/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")