[SPARK-59185][SQL] Derive a StartsWith prefix filter from leading-literal LIKE patterns

python/pyspark/sql/datasource.py

@@ -320,15 +320,19 @@ class Filter(ABC):
     | `a like 'abc%'`     | `StringStartsWith(("a",), "abc")`          |
     | `a like '%abc'`     | `StringEndsWith(("a",), "abc")`            |
     | `a like '%abc%'`    | `StringContains(("a",), "abc")`            |
+    | `a like 'c%c%'`     | `StringStartsWith(("a",), "c")`            |
     +---------------------+--------------------------------------------+
 
+    For a `LIKE` pattern with a leading literal that is not one of the simple forms above
+    (e.g. `a like 'c%c%'`), the leading literal is pushed as a `StringStartsWith` prefix
+    filter while the `LIKE` itself is retained and evaluated by Spark.
+
     Unsupported filters
     - `a = b`
     - `f(a, b) = 1`
     - `a % 2 = 1`
     - `a[0] = 1`
     - `a < 0 or a > 1`
-    - `a like 'c%c%'`
     - `a ilike 'hi'`
     - `a = 'hi' collate zh`
     """

python/pyspark/sql/tests/test_python_datasource.py

@@ -1086,7 +1086,6 @@ def test_unsupported_filter(self):
         self._check_filters("int", "(0 < x and x < 1) or x = 2", [])
         self._check_filters("int", "x % 5 = 1", [])
         self._check_filters("array<int>", "x[0] = 1", [])
-        self._check_filters("string", "x like 'a%a%'", [])
         self._check_filters("string", "x ilike 'a'", [])
         self._check_filters("string", "x = 'a' collate zh", [])
 
@@ -1129,6 +1128,10 @@ def test_filter_type(self):
         self._check_filters(
             "string", "x like 'a%b'", [StringStartsWith(("x",), "a"), StringEndsWith(("x",), "b")]
         )
+        # A leading-literal multi-wildcard pattern is not fully simplified, but its leading
+        # literal is derived as a StringStartsWith prefix filter; the LIKE stays as the exact
+        # (non-pushed) residual.
+        self._check_filters("string", "x like 'a%a%'", [StringStartsWith(("x",), "a")])
         self._check_filters("int", "x in (1, 2)", [In(("x",), [1, 2])])
 
     def test_filter_nested_column(self):

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala

@@ -813,6 +813,11 @@ object LikeSimplification extends Rule[LogicalPlan] with PredicateHelper {
   private val contains = "%+([^_%]+)%+".r
   private val equalTo = "([^_%]*)".r
 
+  // Marks a residual `Like` that `derivePrefixStartsWith` has already guarded with a leading
+  // `StartsWith`, so the rule does not re-wrap it on later fixed-point iterations (which would
+  // otherwise loop and break the batch's idempotence). Tags are ignored by `fastEquals`.
+  private[sql] val LIKE_PREFIX_GUARDED = TreeNodeTag[Unit]("likePrefixStartsWithAdded")
+
   private def simplifyLike(
       input: Expression, pattern: String, escapeChar: Char = '\\'): Option[Expression] = {
     if (pattern.contains(escapeChar)) {
@@ -863,6 +868,43 @@ object LikeSimplification extends Rule[LogicalPlan] with PredicateHelper {
     }
   }
 
+  // For a leading-literal pattern that `simplifyLike` leaves as a full `Like` (e.g. 'a%b%'),
+  // derive the necessary condition `StartsWith(input, <leading literal>)` and keep the `Like`
+  // as the exact residual: `StartsWith(input, prefix) && (input LIKE pattern)`. `StartsWith` is
+  // placed first so the cheap check short-circuits the regex, and it can be pushed to data
+  // sources (e.g. Parquet prunes row groups on `StringStartsWith`) while the `Like` re-checks
+  // the match exactly.
+  //
+  // Restricted to collations with binary equality: only then do the `Like` regex match and
+  // `StartsWith` agree byte-for-byte, so `StartsWith(prefix)` is a sound necessary condition of
+  // the `Like` (under e.g. UTF8_LCASE the two matchers can disagree, risking a false negative),
+  // and only then does `StringStartsWith` push down. The residual `Like` is tagged so the rule
+  // stays idempotent under the fixed-point batch.
+  private def derivePrefixStartsWith(
+      input: Expression,
+      pattern: String,
+      escapeChar: Char,
+      like: Expression): Option[Expression] = {
+    val binaryCollation = input.dataType match {
+      case st: StringType => st.supportsBinaryEquality
+      case _ => false
+    }
+    if (!binaryCollation || !CollapseProject.isCheap(input) || pattern.contains(escapeChar) ||
+        like.containsTag(LIKE_PREFIX_GUARDED)) {
+      None
+    } else {
+      val prefix = pattern.takeWhile(c => c != '%' && c != '_')
+      if (prefix.isEmpty || prefix.length == pattern.length) {
+        // No leading literal (pattern starts with a wildcard), or no wildcard at all (the
+        // latter is already turned into `EqualTo` by `simplifyLike`).
+        None
+      } else {
+        like.setTagValue(LIKE_PREFIX_GUARDED, ())
+        Some(And(StartsWith(input, Literal.create(prefix, input.dataType)), like))
+      }
+    }
+  }
+
   private def simplifyMultiLike(
       child: Expression, patterns: Seq[UTF8String], multi: MultiLikeBase): Expression = {
     val (remainPatternMap, replacementMap) =
@@ -898,7 +940,10 @@ object LikeSimplification extends Rule[LogicalPlan] with PredicateHelper {
         // If pattern is null, return null value directly, since "col like null" == null.
         Literal(null, BooleanType)
       } else {
-        simplifyLike(input, pattern.toString, escapeChar).getOrElse(l)
+        val patternStr = pattern.toString
+        simplifyLike(input, patternStr, escapeChar)
+          .orElse(derivePrefixStartsWith(input, patternStr, escapeChar, l))
+          .getOrElse(l)
       }
     case l @ LikeAll(child, patterns) if CollapseProject.isCheap(child) =>
       simplifyMultiLike(child, patterns, l)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LikeSimplificationSuite.scala

@@ -35,6 +35,16 @@ class LikeSimplificationSuite extends PlanTest {
         LikeSimplification) :: Nil
   }
 
+  // Runs LikeSimplification to a fixed point alongside the operator-optimization rules that can
+  // rebuild the residual Like (and drop its idempotence tag). Deliberately excludes the rules that
+  // collapse an empty LocalRelation (PropagateEmptyRelation / ConvertToLocalRelation), which the
+  // full optimizer would apply -- they would erase the Filter and the derived predicate we assert.
+  object OptimizeWithFixedPoint extends RuleExecutor[LogicalPlan] {
+    val batches =
+      Batch("Operator Optimization", FixedPoint(100),
+        LikeSimplification, BooleanSimplification, PruneFilters) :: Nil
+  }
+
   val testRelation = LocalRelation($"a".string)
 
   test("simplify Like into StartsWith") {
@@ -312,6 +322,72 @@ class LikeSimplificationSuite extends PlanTest {
     comparePlans(Optimize.execute(originalQuery), originalQuery)
   }
 
+  test("derive StartsWith prefix guard for leading-literal LIKE 'a%b%'") {
+    val originalQuery = testRelation.where($"a" like "a%b%")
+    val optimized = Optimize.execute(originalQuery.analyze)
+    val correctAnswer = testRelation
+      .where(StartsWith($"a", "a") && ($"a" like "a%b%"))
+      .analyze
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("derive StartsWith prefix guard with a multi-char prefix and multiple wildcards") {
+    val originalQuery = testRelation.where($"a" like "ab%cd%ef")
+    val optimized = Optimize.execute(originalQuery.analyze)
+    val correctAnswer = testRelation
+      .where(StartsWith($"a", "ab") && ($"a" like "ab%cd%ef"))
+      .analyze
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("derive StartsWith prefix guard when the pattern uses '_' wildcards") {
+    val originalQuery = testRelation.where($"a" like "a_b%")
+    val optimized = Optimize.execute(originalQuery.analyze)
+    val correctAnswer = testRelation
+      .where(StartsWith($"a", "a") && ($"a" like "a_b%"))
+      .analyze
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("no StartsWith prefix guard when the pattern has no leading literal") {
+    val originalQuery = testRelation.where($"a" like "%b%c%").analyze
+    comparePlans(Optimize.execute(originalQuery), originalQuery)
+  }
+
+  test("no StartsWith prefix guard when the pattern contains the escape char") {
+    val originalQuery = testRelation.where($"a" like "a\\%b%").analyze
+    comparePlans(Optimize.execute(originalQuery), originalQuery)
+  }
+
+  test("no StartsWith prefix guard for non-binary collation") {
+    val relation = LocalRelation(AttributeReference("a", StringType("UTF8_LCASE"))())
+    val lcase = StringType("UTF8_LCASE")
+    val originalQuery =
+      relation.where(Like(relation.output.head, Literal.create("a%b%", lcase), '\\')).analyze
+    comparePlans(Optimize.execute(originalQuery), originalQuery)
+  }
+
+  test("SPARK-59185: no StartsWith prefix guard when the child is not a cheap expression") {
+    // The derivation duplicates the child (`StartsWith(child, ..) && (child LIKE ..)`). Mirroring
+    // the SPARK-40228 gate on the multiLike rules, it must not fire for a non-cheap child:
+    // duplicating one re-evaluates it (and a nondeterministic child would yield two values).
+    val originalQuery = testRelation.where($"a".substring(1, 5) like "a%b%").analyze
+    comparePlans(Optimize.execute(originalQuery), originalQuery)
+  }
+
+  test("SPARK-59185: prefix guard derivation is idempotent under a fixed-point batch") {
+    // Not a Once batch: running to a fixed point alongside BooleanSimplification/PruneFilters
+    // exercises the LIKE_PREFIX_GUARDED tag. Were it lost, LikeSimplification would re-derive the
+    // StartsWith on every iteration -- double-wrapping it and eventually tripping the batch's
+    // max-iterations check -- instead of converging to a single guarded rewrite.
+    val originalQuery = testRelation.where($"a" like "a%b%").analyze
+    val optimized = OptimizeWithFixedPoint.execute(originalQuery)
+    val correctAnswer = testRelation
+      .where(StartsWith($"a", "a") && ($"a" like "a%b%"))
+      .analyze
+    comparePlans(optimized, correctAnswer)
+  }
+
   // scalastyle:off nonascii
   test("SPARK-59063: LikeSimplification preserves LIKE semantics under non-binary collation") {
     // Under UTF8_LCASE, StartsWith/EndsWith are collation-aware, so a single code point

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala

@@ -113,6 +113,13 @@ abstract class ParquetFilterSuite extends ParquetTest with SharedSparkSession {
       checker: (DataFrame, Seq[Row]) => Unit,
       expected: Seq[Row]): Unit
 
+  /**
+   * Returns the source `Filter`s pushed down to the Parquet scan for `query`. This hides the
+   * V1/V2 difference in where pushdown happens (V1: physical planning via
+   * `DataSourceStrategy.selectFilters`; V2: optimizer, read back from `ParquetScan.pushedFilters`).
+   */
+  protected def getPushedDownFilters(query: DataFrame): Seq[sources.Filter]
+
   private def checkFilterPredicate
       (predicate: Predicate, filterClass: Class[_ <: FilterPredicate], expected: Seq[Row])
       (implicit df: DataFrame): Unit = {
@@ -1779,6 +1786,52 @@ abstract class ParquetFilterSuite extends ParquetTest with SharedSparkSession {
     }
   }
 
+  test("filter pushdown - leading-literal LIKE derives a StartsWith prefix filter") {
+    import testImplicits._
+    // A multi-wildcard pattern with a leading literal (e.g. 'ab%cd%') is not rewritten to a
+    // single StartsWith/EndsWith/Contains, but LikeSimplification also derives the necessary
+    // condition StartsWith(<leading literal>), which pushes down and prunes row groups whose
+    // min/max cannot contain the prefix. The digit-string data below has no value starting with
+    // the alphabetic prefix, so canDrop() removes every row group.
+    Seq(
+      "value like 'ab%cd%'",   // leading literal 'ab'
+      "value like 'ab%cd%ef'", // leading literal 'ab', trailing literal 'ef'
+      "value like 'a_b%'"      // leading literal 'a' before an '_' wildcard
+    ).foreach { filter =>
+      testStringPredicate(
+        spark.range(1024).map(_.toString).toDF(),
+        filter,
+        shouldFilterOut = true,
+        enableDictionary = false)
+    }
+  }
+
+  test("SPARK-59185: leading-literal LIKE pushes StringStartsWith and keeps the residual LIKE") {
+    import testImplicits._
+    withSQLConf(
+      SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED.key -> "true",
+      // Keep pushed filters clean (constraint inference would add IsNotNull), as other tests do.
+      SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> InferFiltersFromConstraints.ruleName) {
+      withTempPath { dir =>
+        val path = dir.getCanonicalPath
+        // "abXcdY" matches ab%cd%; "abZZZ" hits the 'ab' prefix but fails the rest; "zzz" has no
+        // prefix. Written as one file so the residual, not row-group pruning, does the rejection.
+        Seq("abXcdY", "abZZZ", "zzz").toDF("value").write.parquet(path)
+        val query = spark.read.parquet(path).where("value like 'ab%cd%'")
+
+        // (a) The derived leading-literal condition reaches Parquet as a StringStartsWith filter.
+        val pushed = getPushedDownFilters(query)
+        assert(pushed.contains(sources.StringStartsWith("value", "ab")),
+          s"expected StringStartsWith('value', 'ab') among pushed filters, got: $pushed")
+
+        // (b) The residual LIKE still filters: StartsWith('ab') is only a necessary condition, so a
+        // prefix hit that fails the rest of the pattern ("abZZZ") must be rejected -- a bug that
+        // dropped the residual LIKE and kept only StartsWith would wrongly return it.
+        checkAnswer(query, Row("abXcdY"))
+      }
+    }
+  }
+
   test("SPARK-17091: Convert IN predicate to Parquet filter push-down") {
     val schema = StructType(Seq(
       StructField("a", IntegerType, nullable = false)
@@ -2937,6 +2990,19 @@ class ParquetV1FilterSuite extends ParquetFilterSuite {
       }
     }
   }
+
+  override protected def getPushedDownFilters(query: DataFrame): Seq[sources.Filter] = {
+    var maybeRelation: Option[HadoopFsRelation] = None
+    val analyzedPredicate = query.queryExecution.optimizedPlan.collect {
+      case PhysicalOperation(_, filters,
+          LogicalRelationWithTable(relation: HadoopFsRelation, _)) =>
+        maybeRelation = Some(relation)
+        filters
+    }.flatten
+    maybeRelation
+      .map(DataSourceStrategy.selectFilters(_, analyzedPredicate)._2)
+      .getOrElse(Seq.empty)
+  }
 }
 
 @ExtendedSQLTest
@@ -3000,6 +3066,13 @@ class ParquetV2FilterSuite extends ParquetFilterSuite {
     }
   }
 
+  override protected def getPushedDownFilters(query: DataFrame): Seq[sources.Filter] = {
+    query.queryExecution.optimizedPlan.collectFirst {
+      case PhysicalOperation(_, _, ExtractV2Scan(scan: ParquetScan)) =>
+        scan.pushedFilters.toImmutableArraySeq
+    }.getOrElse(Seq.empty)
+  }
+
   test("SPARK-36889: Respect disabling of filters pushdown for DSv2 by explain") {
     import testImplicits._
     withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false") {