From ac56b5c505dc97241757a0ab16619a68ad7376dc Mon Sep 17 00:00:00 2001 From: David Mollitor Date: Wed, 2 Sep 2026 18:28:00 +0000 Subject: [PATCH] [SPARK-59185][SQL] Derive a StartsWith prefix filter from leading-literal LIKE patterns `LikeSimplification` rewrites simple `LIKE` patterns into cheaper predicates (`'A%'` -> `StartsWith`, `'%B'` -> `EndsWith`, `'A%B'` -> length guard + `StartsWith` + `EndsWith`, `'%B%'` -> `Contains`, exact -> `EqualTo`). Multi-wildcard patterns with a leading literal that match none of those shapes -- e.g. `'A%B%'`, `'AB%CD%EF'`, `'A_B%'` -- fall through unchanged and stay a full regex `Like`, so the data source receives no predicate and the per-row regex runs on every row. This derives the leading literal `A` as the necessary condition `StartsWith(col, A)` and keeps the original `LIKE` as the exact residual: col LIKE 'A%B%' ==> StartsWith(col, A) && (col LIKE 'A%B%') `StartsWith` is placed first so the cheap check short-circuits the regex, and it is a predicate the existing pushdown path understands: on UTF8_BINARY it translates to `StringStartsWith`, which prunes Parquet row groups via min/max. Results are unchanged -- `StartsWith(A)` is implied by `LIKE 'A%...'` and the exact `LIKE` is retained as the residual. The derivation is restricted to binary-equality collations (`supportsBinaryEquality`): under a collation-aware collation the `Like` regex match (Java regex case flags) and `StartsWith` (`CollationSupport`) can disagree, so `StartsWith(A)` would not be a sound necessary condition; and `StringStartsWith` only pushes down for UTF8_BINARY. Only `StartsWith` is derived from the leading literal -- Parquet's `StringEndsWith` and `StringContains` do not prune -- and the `LikeAll`/`LikeAny` paths are unchanged. A `TreeNodeTag` on the residual `Like` keeps the rule idempotent under the fixed-point batch. This also updates the Python data source filter-pushdown test and the `DataSource.pushFilters` docstring, which previously documented such patterns as pushing no filters. Generated-by: Claude Opus 4.8 --- python/pyspark/sql/datasource.py | 6 +- .../sql/tests/test_python_datasource.py | 5 +- .../sql/catalyst/optimizer/expressions.scala | 47 +++++++++++- .../optimizer/LikeSimplificationSuite.scala | 76 +++++++++++++++++++ .../parquet/ParquetFilterSuite.scala | 73 ++++++++++++++++++ 5 files changed, 204 insertions(+), 3 deletions(-) diff --git a/python/pyspark/sql/datasource.py b/python/pyspark/sql/datasource.py index 2afdd2ebc260a..097bb789de778 100644 --- a/python/pyspark/sql/datasource.py +++ b/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` """ diff --git a/python/pyspark/sql/tests/test_python_datasource.py b/python/pyspark/sql/tests/test_python_datasource.py index 5bb8e9df1e3b2..6732d448549ff 100644 --- a/python/pyspark/sql/tests/test_python_datasource.py +++ b/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", "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): diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala index cdc0444d74cc1..8a2dc0fd6754a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala +++ b/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, )` 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) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LikeSimplificationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LikeSimplificationSuite.scala index 8b142f0c53d75..5007168d122f3 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LikeSimplificationSuite.scala +++ b/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 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index 9b851134fc0b2..8d8a27bf02a70 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/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(), 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") {