Conversation
|
cc @szehon-ho |
uros-b
left a comment
There was a problem hiding this comment.
Looks clean to me, thank you @pan3793! But yeah let's definitely add @szehon-ho who has more context in PartitionPredicate
dongjoon-hyun
left a comment
There was a problem hiding this comment.
+1, LGTM.
A few suggestions:
getPartitionPredicateSchemahas a fourth caller,GroupBasedRowLevelOperationScanPlanning(group-based UPDATE/MERGE/DELETE scans), which the description does not mention. Could you mention it and add a mixed-partitioning MERGE or UPDATE test?- In "filter on the source column of a bucket transform stays post-scan",
exists(_.isInstanceOf[FilterExec])is a weak check. Could you assert that theFilterExeccondition referencesid? - nit: "mixed partitioning: no identity transform" uses a bucket-only table, so it is not mixed.
- nit: the warning in
PartitionPredicateImpl.applylists non-identity fields (e.g.bucket(4, id)) as partition fields a filter could reference.
| * flattened dotted name (e.g. `"s.tz"`) for nested fields. | ||
| * (e.g. `Seq("s", "tz")`) for an identity transform, or the transform's | ||
| * description (e.g. `Seq("bucket(4, id)")`) otherwise. | ||
| * @param attrRef the [[AttributeReference]] a filter can reference, for an identity transform. |
There was a problem hiding this comment.
nit: we can say 'for now Spark doesnt support'. it was in the plan but never implemented yet
| case p: Predicate if p.name().equals("IS_NOT_NULL") => true | ||
| case p: Predicate if p.name().equals("ALWAYS_TRUE") => true | ||
| case _ => false | ||
| predicates.flatMap(splitAnd).forall { p => |
There was a problem hiding this comment.
suggestion:
def supportsPredicates(predicates: Array[Predicate]): Boolean = {
predicates.flatMap(splitAnd).forall { p =>
(p.name(), p.children().toSeq) match {
case ("=" | "<=>", Seq(_: NamedReference, _: LiteralValue[_])) => true
case ("IS_NULL" | "IS_NOT_NULL", Seq(_: NamedReference)) => true
case ("ALWAYS_TRUE", _) => true
case _ => false
}
}
}
…g() to the wrapped table The read relation of a row-level rewrite (UPDATE, MERGE, group-based DELETE) wraps the table in RowLevelOperationTable, which did not override partitioning(), so the default empty array was returned. Since SPARK-55596 (4.2.0) PushDownUtils.getPartitionPredicateSchema reads it for those scans, found no transforms, and never derived a PartitionPredicate for any row-level operation. Delegate partitioning() to the wrapped table. The in-memory row-level fixture now pushes V2 predicates iteratively, and a group-based UPDATE test checks that the second-pass PartitionPredicate reaches the scan and that only the matching partitions are replaced. Assisted-by: Claude Fable 5.1
…a mixed partitioning PushDownUtils.getPartitionPredicateSchema returned a schema only when every transform in Table.partitioning() is an identity transform, so a table partitioned by e.g. dt (identity) and bucket(16, user_id) never received a PartitionPredicate, and a Catalyst-only filter on dt such as the cast(dt AS DATE) = DATE'...' produced by type coercion could not prune partitions in the static pass, via DPP, or in a metadata-only DELETE. The schema now has one field per transform, in partitioning order. Identity fields carry an attribute a filter can reference; other fields have none but keep their ordinal, so a predicate still binds against the full partition key and the connector contract is unchanged. A filter on the source column of a non-identity transform stays a data filter. A partitioning with no identity transform still yields no schema. The in-memory V2 filter test table now accepts only column-vs-literal predicates and returns anything else, e.g. a predicate over a cast, as a real connector would. Assisted-by: Claude Fable 5.1 (cherry picked from commit d4bfb94)
- UPDATE and MERGE tests on a mixed partitioning through the row-level scan - assert the post-scan filter references the bucket source column only - rename the bucket-only test, tighten supportsPredicates, list only identity fields in the unmatched-reference warning Assisted-by: Claude Fable 5.1
d4bfb94 to
d0d6b52
Compare
|
Thanks, all four addressed in d0d6b52.
|
szehon-ho
left a comment
There was a problem hiding this comment.
LGTM. The production row-level path looks correct. I left one non-blocking, test-only follow-up.
| override def pushPredicates(predicates: Array[Predicate]): Array[Predicate] = { | ||
| val (accepted, returned) = predicates.partition { | ||
| case _: PartitionPredicate => acceptPartitionPredicates | ||
| case p => refsOnlyPartCols(p) && InMemoryTableWithV2Filter.supportsPredicates(Array(p)) |
There was a problem hiding this comment.
Test-only follow-up: partCols includes references from all transforms. For (dep, bucket(4, pk)), this may accept pk = 4 and evaluate it against the bucket-value slot. This does not affect production code or the mixed-partitioning tests in this PR, but could make the fixture model ordinary filter pushdown incorrectly. Should refsOnlyPartCols use only identity transforms?
| override def columns: Array[Column] = table.columns() | ||
| override def capabilities: util.Set[TableCapability] = table.capabilities | ||
| override def constraints(): Array[Constraint] = table.constraints() | ||
| override def partitioning(): Array[Transform] = table.partitioning() |
There was a problem hiding this comment.
Good catch. Row-level rewrites replace the original relation table with this wrapper, so without this delegation getPartitionPredicateSchema sees the default empty partitioning and disables second-pass PartitionPredicate pushdown. This enables UPDATE/MERGE/DELETE, including mixed partitioning.
Depends on #58756 (SPARK-59457), included here as the first commit.
What changes were proposed in this pull request?
PushDownUtils.getPartitionPredicateSchemareturned a schema only when every transform inTable.partitioning()is an identity transform. It now accepts any partitioning with at leastone identity transform: identity fields can be referenced by a
PartitionPredicate, otherfields keep their ordinal but are never referenced, so a filter on the source column of a
non-identity transform stays a data filter.
The connector contract is unchanged:
PartitionPredicate.evalstill receives the fullpartition key, and
references()still reports ordinals intoTable.partitioning().This applies to all users of the schema: the static second pass, runtime filter pushdown
(DPP and scalar subqueries), the metadata-only DELETE rewrite, and the scan of a group-based
UPDATE, MERGE or DELETE.
The in-memory V2 filter test table now accepts only predicates of the shape it can evaluate,
a column against a literal, and returns anything else, e.g. a predicate over a cast, as a
real connector would.
Why are the changes needed?
A table partitioned by, for example,
dt STRING(identity) andbucket(16, user_id)receivedno
PartitionPredicateat all. A filter such asdt = DATE'2026-09-01'is analyzed ascast(dt AS DATE) = DATE'2026-09-01'; a connector that does not evaluate casts returns itfrom the first pass (and without ANSI mode it is not translatable at all), so only a
PartitionPredicatecan prune with it. On such a table it could not prune partitions ineither the static or the runtime path, and could not drive a metadata-only DELETE, while the
same filter on an identity-only table can. Mixed partitionings are the common case for
connectors that support partition transforms.
The all-identity check was raised in the SPARK-55596 review (#54459) and kept only because
the partition-key contract was still open then. The contract that shipped (full key,
partitioning order) is what makes lifting the check safe.
Does this PR introduce any user-facing change?
No. A connector that opts into iterative pushdown and has a mixed partitioning now receives
PartitionPredicates over its identity fields; theevalcontract is unchanged.How was this patch tested?
New tests on a partitioning of one identity column plus a bucket transform: a predicate on
the identity field is pushed and prunes in the static pass, via DPP, in a metadata-only
DELETE, and in the scan of a group-based UPDATE and MERGE, including the
dt STRINGvsDATEliteral case whose type-coercion cast the source cannot evaluate; an identity fieldplaced after the bucket transform binds to its own ordinal; a filter on the bucket source
column is not turned into a predicate; a bucket-only partitioning yields no predicate; the
predicate survives Java and Kryo round-trips.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5.1