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

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)) {
@@ -846,6 +851,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 || 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) =
@@ -881,7 +923,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

@@ -310,6 +310,57 @@ 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("prefix guard derivation is idempotent") {
+    val originalQuery = testRelation.where($"a" like "a%b%").analyze
+    val once = Optimize.execute(originalQuery)
+    comparePlans(Optimize.execute(once), once)
+  }
+
   // scalastyle:off nonascii
   test("LikeSimplification with emojis") {
     val originalQuery =

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

@@ -1751,6 +1751,26 @@ 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-17091: Convert IN predicate to Parquet filter push-down") {
     val schema = StructType(Seq(
       StructField("a", IntegerType, nullable = false)