Repository navigation
[SPARK-59721][SQL] Ignore reported partitioning and ordering that reference unresolvable columns - #58979
[SPARK-59721][SQL] Ignore reported partitioning and ordering that reference unresolvable columns#58979anuragmantri wants to merge 8 commits into
Conversation
|
@peter-toth - Could you please provide a review |
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for making a PR, @anuragmantri. The direction (degrading to unknown partitioning instead of failing, which makes the sequenceToOption fallback in V2ScanPartitioningAndOrdering.partitioning() effective) looks reasonable to me. I left a few comments.
-
PR title:
Keep storage-partitioned joinlooks misleading. With this PR, the partitioning is degraded toUnknownPartitioning, so SPJ is not applied; only the query stops failing. Could you revise the title, e.g.,Fall back to unknown partitioning instead of failing when a reported partition key cannot be resolved? -
Silent fallback: An unresolvable partition key usually indicates a connector bug. Could you add a
logWarninginV2ScanPartitioningAndOrdering.partitioning()when theKeyGroupedPartitioningis dropped, like the existing warning for non-KeyGroupedPartitioning? Otherwise, it will be hard to notice why SPJ is not applied. -
End-to-end test: The user-facing change (the query now succeeds) is not covered. Could you add a test in
KeyGroupedPartitioningSuitewhere a table reports an unresolvable partition key, verifying that the query succeeds and SPJ is not applied?
| } | ||
| case ref: FieldReference => | ||
| Some(resolveRef[NamedExpression](ref, query)) | ||
| resolveRefOpt[NamedExpression](ref, query) |
There was a problem hiding this comment.
This changes the error of toCatalyst/toCatalystOrdering too, because toCatalyst is toCatalystOpt(...).getOrElse(throw _LEGACY_ERROR_TEMP_3054). The impact is broader than toCatalystOrdering mentioned in the PR description:
DistributionAndOrderingUtils(write path): the clustering ofrequiredDistributionviatoCatalystandrequiredOrderingviatoCatalystOrderingV2ScanPartitioningAndOrdering.ordering():scan.outputOrdering()
For writes, when a connector requires a non-existent column, the error changes from _LEGACY_ERROR_TEMP_1137 (Unable to resolve x given [a,b]) to _LEGACY_ERROR_TEMP_3054 (x is not currently supported), which is misleading.
Could you keep the specific error for the throwing APIs? For example, a private helper with a failOnUnresolved flag which toCatalyst/toCatalystOrdering use, while only the *Opt variants return None. Note that adding a FieldReference case to toCatalyst alone is not enough because SortValue recurses into toCatalystOpt for its child.
There was a problem hiding this comment.
Good catch, thanks. Fixed as you suggested: toCatalyst/toCatalystOpt and toCatalystTransformOpt now go through private toCatalystCore/toCatalystTransformCore helpers that take a failOnUnresolved flag. toCatalyst (and so toCatalystOrdering) passes true and keeps throwing _LEGACY_ERROR_TEMP_1137, while the *Opt variants pass false and return None. The flag is threaded through the SortValue child and the NamedTransform args, so the recursion you pointed out keeps the specific error too. resolveRef itself is unchanged.
The toCatalystOrdering test now asserts _LEGACY_ERROR_TEMP_1137, and I added a direct toCatalyst test for an unresolvable FieldReference. I also removed the side-effect paragraph from the PR description since it no longer applies.
| * query) must use this, not `resolveRef` -- calling the throwing form there defeats that | ||
| * fallback entirely. | ||
| */ | ||
| def resolveRefOpt[T <: NamedExpression](ref: NamedReference, plan: LogicalPlan): Option[T] = { |
There was a problem hiding this comment.
nit. This scaladoc describes the caller-side details at length. A one or two line description would be enough, e.g., Non-throwing variant of resolveRef. Returns None if the reference cannot be resolved. In addition, can we make this private[sql]?
There was a problem hiding this comment.
Done. Used your wording for the scaladoc and made it private[sql].
| } | ||
| } | ||
| for { | ||
| catalystArgs <- sequenceToOption(args.map(toCatalystOpt(_, query, funCatalogOpt))) |
There was a problem hiding this comment.
This switches toCatalyst to toCatalystOpt for the arguments, so not only unresolvable references but also nested transforms whose function cannot be loaded now return None instead of throwing. It seems intended, but could you mention it in the PR description? Also, unsupported expression types still throw from case _ => in toCatalystOpt, so the graceful degradation is partial.
There was a problem hiding this comment.
Yes, that's intended. I added both points to the PR description. On the throwing path, a nested transform whose function can't be loaded still fails with _LEGACY_ERROR_TEMP_3054, but the message now names the outer transform instead of the inner one.
| assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently supported")) | ||
| } | ||
|
|
||
| test("toCatalystOpt returns None, not throw, for an unresolvable FieldReference") { |
There was a problem hiding this comment.
Please add the JIRA ID prefix to the new test names, e.g., test("SPARK-59721: ...").
| assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"), plan).isEmpty) | ||
| } | ||
|
|
||
| test("toCatalystOpt still resolves an existing FieldReference") { |
There was a problem hiding this comment.
nit. This test case looks trivial. Do we need it?
| } | ||
|
|
||
| test("toCatalystTransformOpt returns None for an unresolvable BucketTransform ref") { | ||
| assert(V2ExpressionUtils.toCatalystTransformOpt(Expressions.bucket(4, "missing"), plan).isEmpty) |
There was a problem hiding this comment.
Since funCatalogOpt is None here (and in the years test below), the result would be None even if the reference were resolvable. It distinguishes the old behavior (exception), but could you pass a function catalog so that the test proves None comes from the resolution failure?
There was a problem hiding this comment.
Good point. The bucket and years tests now pass a function catalog: a small UnboundFunction registered under both names in an InMemoryCatalog. Each test also asserts that the same transform on the existing column a converts, so the None for missing can only come from the resolution failure.
|
Gentle ping, @anuragmantri . |
anuragmantri
left a comment
There was a problem hiding this comment.
Thanks for the review, @dongjoon-hyun. I addressed all comments in the latest commit.
- PR title: Updated to your suggestion.
- Silent fallback: Added a
logWarninginV2ScanPartitioningAndOrdering.partitioning()when the reportedKeyGroupedPartitioningis dropped because its keys reference columns that cannot be resolved. The message names those columns. Drops for other reasons, like a missing function catalog or function, keep their existing behavior, since those aren't connector bugs and a missing function already has its own warning inloadV2FunctionOpt. - End-to-end test: Added
SPARK-59721: an unresolvable reported partition key falls back to unknown partitioninginKeyGroupedPartitioningSuite. The in-memory tables always report their real partition columns, and a column-pruned key still resolves against the full relation, so neither can produce this case. The test registers a small catalog whose scan reports a key on a column namedmissing, joins it with a regular table, and checks that the query returns the right rows, that the scan's partitioning is dropped, that a shuffle is planned, and that the warning namesmissing.
| assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"), plan).isEmpty) | ||
| } | ||
|
|
||
| test("toCatalystOpt still resolves an existing FieldReference") { |
| assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently supported")) | ||
| } | ||
|
|
||
| test("toCatalystOpt returns None, not throw, for an unresolvable FieldReference") { |
| } | ||
| case ref: FieldReference => | ||
| Some(resolveRef[NamedExpression](ref, query)) | ||
| resolveRefOpt[NamedExpression](ref, query) |
There was a problem hiding this comment.
Good catch, thanks. Fixed as you suggested: toCatalyst/toCatalystOpt and toCatalystTransformOpt now go through private toCatalystCore/toCatalystTransformCore helpers that take a failOnUnresolved flag. toCatalyst (and so toCatalystOrdering) passes true and keeps throwing _LEGACY_ERROR_TEMP_1137, while the *Opt variants pass false and return None. The flag is threaded through the SortValue child and the NamedTransform args, so the recursion you pointed out keeps the specific error too. resolveRef itself is unchanged.
The toCatalystOrdering test now asserts _LEGACY_ERROR_TEMP_1137, and I added a direct toCatalyst test for an unresolvable FieldReference. I also removed the side-effect paragraph from the PR description since it no longer applies.
| * query) must use this, not `resolveRef` -- calling the throwing form there defeats that | ||
| * fallback entirely. | ||
| */ | ||
| def resolveRefOpt[T <: NamedExpression](ref: NamedReference, plan: LogicalPlan): Option[T] = { |
There was a problem hiding this comment.
Done. Used your wording for the scaladoc and made it private[sql].
| } | ||
| } | ||
| for { | ||
| catalystArgs <- sequenceToOption(args.map(toCatalystOpt(_, query, funCatalogOpt))) |
There was a problem hiding this comment.
Yes, that's intended. I added both points to the PR description. On the throwing path, a nested transform whose function can't be loaded still fails with _LEGACY_ERROR_TEMP_3054, but the message now names the outer transform instead of the inner one.
| } | ||
|
|
||
| test("toCatalystTransformOpt returns None for an unresolvable BucketTransform ref") { | ||
| assert(V2ExpressionUtils.toCatalystTransformOpt(Expressions.bucket(4, "missing"), plan).isEmpty) |
There was a problem hiding this comment.
Good point. The bucket and years tests now pass a function catalog: a small UnboundFunction registered under both names in an InMemoryCatalog. Each test also asserts that the same transform on the existing column a converts, so the None for missing can only come from the resolution failure.
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing the previous comments, @anuragmantri. The failOnUnresolved helper keeps the specific errors for the throwing APIs, and the new warning and end-to-end test look good to me.
I left a few more comments. In short, the fallback is still partial and can be silent in some cases.
- Silent fallback: A nested transform argument (e.g.,
f(g(missing))) degrades toNonewithout the new warning, becauseTransform.references()returns only top-level references. A nested transform whose function cannot be loaded is now dropped without any log too, while it failed before. - Queries which still fail: A nested field key (e.g.,
s.missing) still throwsFIELD_NOT_FOUNDfromresolveRefOpt, and a scan that also reportsoutputOrdering()on the same column still fails inordering(). - Test coverage: The throwing path through
BucketTransformrefs andNamedTransformargs (used by the write path) and transform keys in the end-to-end test are not covered. - Minor: The fallback decision and the warning are computed separately, and the warning can list duplicate column names.
| kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog)) | ||
| .toImmutableArraySeq) | ||
| if (partitioning.isEmpty) { | ||
| val unresolvedRefs = kgp.keys().flatMap(_.references()).filter { ref => |
There was a problem hiding this comment.
Transform.references() returns only the top-level NamedReference arguments for ApplyTransform (arguments.collect { case named: NamedReference => named }). Since this PR routes NamedTransform args through the non-throwing path, a key like f(g(missing)) now degrades to None while unresolvedRefs is empty, so no warning is logged at all. Before this PR, the query failed with _LEGACY_ERROR_TEMP_1137. Could you collect the references recursively so that this case is not silently ignored?
There was a problem hiding this comment.
Good catch. The check now collects the key references recursively through children(), so a key like f(g(missing)) is caught. It also runs before toCatalystOpt now, so the partitioning is dropped and the warning is logged from the same place. I added f(g(missing)) to the end-to-end test.
| import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper | ||
|
|
||
| /** | ||
| * Non-throwing variant of `resolveRef`. Returns `None` if the reference cannot be resolved. |
There was a problem hiding this comment.
BTW, the scaladoc wording that I suggested earlier turns out to be imprecise. resolveRefOpt is not strictly non-throwing: for a multi-part name, LogicalPlan.resolve goes through ExtractValue.applyOrNull, so a key like s.missing (where s is an existing struct column) still throws FIELD_NOT_FOUND, and an ambiguous reference throws as well. So the fallback is still partial for nested partition keys. Could you mention this limitation in the scaladoc and the PR description, or add a test that pins the current behavior?
There was a problem hiding this comment.
You're right, thanks. I updated the scaladoc to say it still throws for a missing nested field or an ambiguous reference. The check in V2ScanPartitioningAndOrdering now treats an AnalysisException from resolveRefOpt as unresolvable, so a key like s.missing falls back with the warning instead of failing the query. resolveRef is unchanged, so the throwing APIs keep FIELD_NOT_FOUND. The end-to-end test covers s.missing on a struct column.
| log"Spark ignores the reported ${MDC(CLASS_NAME, kgp.getClass.getSimpleName)} " + | ||
| log"because the partition key columns cannot be resolved: " + | ||
| log"${MDC(COLUMN_NAMES, unresolvedRefs.map(_.describe()).mkString(", "))}. " + | ||
| log"Storage-partitioned join will not be applied for this scan.") |
There was a problem hiding this comment.
This message implies that the query proceeds without SPJ, but ordering() below still uses the throwing toCatalystOrdering. If a scan also reports outputOrdering() on the same unresolvable column, the query still fails with _LEGACY_ERROR_TEMP_1137. Should we degrade the reported ordering in the same way (log a warning and drop it), or at least clarify this scope in the PR description?
There was a problem hiding this comment.
Good point. ordering() now runs the same column check on the reported ordering. If a column can't be resolved, it logs a warning and drops the reported ordering instead of failing. I added a test where the scan reports an ordering on missing.
| } | ||
| for { | ||
| catalystArgs <- sequenceToOption( | ||
| args.map(toCatalystCore(_, query, funCatalogOpt, failOnUnresolved))) |
There was a problem hiding this comment.
As mentioned in the PR description, a nested transform whose function cannot be loaded now returns None here instead of failing with _LEGACY_ERROR_TEMP_3054. Note that this drop is completely silent when funCatalogOpt is None or loadFunction throws UnsupportedOperationException: loadV2FunctionOpt warns only on NoSuchFunctionException, and the new warning in V2ScanPartitioningAndOrdering is not triggered because all references are resolvable. Since this turns a query failure into a silent loss of SPJ, could you at least add a test for this case so that the new behavior is explicit?
There was a problem hiding this comment.
Added a test: for f(g(a)) where g is not in the function catalog, toCatalystTransformOpt returns None, and toCatalyst still fails with _LEGACY_ERROR_TEMP_3054 naming f(g(a)). The silent drop matches what already happens for a top-level transform whose function can't be loaded (no function catalog, or bind throwing UnsupportedOperationException), so I kept the two consistent.
| assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"), plan).isEmpty) | ||
| } | ||
|
|
||
| test("SPARK-59721: toCatalyst throws the specific resolution error for an unresolvable " + |
There was a problem hiding this comment.
failOnUnresolved is also threaded through the BucketTransform refs and the NamedTransform args, but only FieldReference and SortValue are covered for the throwing path. Could you add cases like toCatalyst(Expressions.bucket(4, "missing"), plan, funCatalog) and toCatalyst(Expressions.years("missing"), plan, funCatalog) expecting _LEGACY_ERROR_TEMP_1137? This is the path that DistributionAndOrderingUtils uses for ClusteredDistribution on the write side.
There was a problem hiding this comment.
Added toCatalyst(Expressions.bucket(4, "missing"), plan, funCatalog) and toCatalyst(Expressions.years("missing"), plan, funCatalog), both asserting _LEGACY_ERROR_TEMP_1137 with checkError. I also switched the existing error assertions in this suite to checkError.
| override def toBatch: Batch = inner.toBatch | ||
| override def description(): String = inner.description() | ||
| override def outputPartitioning(): Partitioning = new KeyGroupedPartitioning( | ||
| Array[Expression](FieldReference("missing")), |
There was a problem hiding this comment.
This end-to-end test covers only a single identity key. Could you also cover a transform key (e.g., bucket(4, missing)), a nested transform key, and a multi-key case where only one key is unresolvable (e.g., [id, missing]) to verify that the whole partitioning is dropped? The nested case would catch the missing warning which I mentioned in V2ScanPartitioningAndOrdering.
There was a problem hiding this comment.
Done. Each test now sets the keys (and ordering) the fixture scan reports, so one table covers all the cases. The end-to-end test now covers a resolvable control key, bucket(4, missing), f(g(missing)), [id, missing], [missing, bucket(4, missing)], and s.missing. For each unresolvable case it checks that the warning names exactly the unresolvable columns, that the whole partitioning is dropped, and that a shuffle is planned. For [id, missing], only missing is listed.
| val partitioning = sequenceToOption( | ||
| kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog)) | ||
| .toImmutableArraySeq) | ||
| if (partitioning.isEmpty) { |
There was a problem hiding this comment.
Not a blocker, but since toCatalystOpt does not tell why it returned None, this block re-resolves every reference to reconstruct the reason, and the two can diverge (e.g., the nested case in my other comment on unresolvedRefs). Have you considered checking the key references (recursively) against relation up front in this rule, and logging the warning and returning None before calling toCatalystOpt? Then the fallback decision and the warning would come from the same place.
There was a problem hiding this comment.
Yes, I changed it that way. The rule now checks the key references recursively against relation first. If any can't be resolved, it logs the warning and uses None without calling toCatalystOpt; otherwise the existing path runs unchanged. The reported ordering uses the same check.
| logWarning( | ||
| log"Spark ignores the reported ${MDC(CLASS_NAME, kgp.getClass.getSimpleName)} " + | ||
| log"because the partition key columns cannot be resolved: " + | ||
| log"${MDC(COLUMN_NAMES, unresolvedRefs.map(_.describe()).mkString(", "))}. " + |
There was a problem hiding this comment.
nit. Could you add .distinct to unresolvedRefs? Otherwise, the same column is listed multiple times when several keys reference it, e.g., identity(missing) and bucket(4, missing) give missing, missing. In addition, the same warning can be logged more than once for a query (e.g., when PlanMerger re-applies this rule to a merged scan), but that's the same as the existing warning below.
There was a problem hiding this comment.
Done, the column names are de-duplicated, and the [missing, bucket(4, missing)] case checks that missing is listed once. Agreed on the repeated logging; I left that as is, like the existing warning.
anuragmantri
left a comment
There was a problem hiding this comment.
Thanks for the second pass, @dongjoon-hyun. I addressed all comments in the latest commit. Details are in the inline replies.
| kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog)) | ||
| .toImmutableArraySeq) | ||
| if (partitioning.isEmpty) { | ||
| val unresolvedRefs = kgp.keys().flatMap(_.references()).filter { ref => |
There was a problem hiding this comment.
Good catch. The check now collects the key references recursively through children(), so a key like f(g(missing)) is caught. It also runs before toCatalystOpt now, so the partitioning is dropped and the warning is logged from the same place. I added f(g(missing)) to the end-to-end test.
| import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper | ||
|
|
||
| /** | ||
| * Non-throwing variant of `resolveRef`. Returns `None` if the reference cannot be resolved. |
There was a problem hiding this comment.
You're right, thanks. I updated the scaladoc to say it still throws for a missing nested field or an ambiguous reference. The check in V2ScanPartitioningAndOrdering now treats an AnalysisException from resolveRefOpt as unresolvable, so a key like s.missing falls back with the warning instead of failing the query. resolveRef is unchanged, so the throwing APIs keep FIELD_NOT_FOUND. The end-to-end test covers s.missing on a struct column.
| log"Spark ignores the reported ${MDC(CLASS_NAME, kgp.getClass.getSimpleName)} " + | ||
| log"because the partition key columns cannot be resolved: " + | ||
| log"${MDC(COLUMN_NAMES, unresolvedRefs.map(_.describe()).mkString(", "))}. " + | ||
| log"Storage-partitioned join will not be applied for this scan.") |
There was a problem hiding this comment.
Good point. ordering() now runs the same column check on the reported ordering. If a column can't be resolved, it logs a warning and drops the reported ordering instead of failing. I added a test where the scan reports an ordering on missing.
| } | ||
| for { | ||
| catalystArgs <- sequenceToOption( | ||
| args.map(toCatalystCore(_, query, funCatalogOpt, failOnUnresolved))) |
There was a problem hiding this comment.
Added a test: for f(g(a)) where g is not in the function catalog, toCatalystTransformOpt returns None, and toCatalyst still fails with _LEGACY_ERROR_TEMP_3054 naming f(g(a)). The silent drop matches what already happens for a top-level transform whose function can't be loaded (no function catalog, or bind throwing UnsupportedOperationException), so I kept the two consistent.
| assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"), plan).isEmpty) | ||
| } | ||
|
|
||
| test("SPARK-59721: toCatalyst throws the specific resolution error for an unresolvable " + |
There was a problem hiding this comment.
Added toCatalyst(Expressions.bucket(4, "missing"), plan, funCatalog) and toCatalyst(Expressions.years("missing"), plan, funCatalog), both asserting _LEGACY_ERROR_TEMP_1137 with checkError. I also switched the existing error assertions in this suite to checkError.
| override def toBatch: Batch = inner.toBatch | ||
| override def description(): String = inner.description() | ||
| override def outputPartitioning(): Partitioning = new KeyGroupedPartitioning( | ||
| Array[Expression](FieldReference("missing")), |
There was a problem hiding this comment.
Done. Each test now sets the keys (and ordering) the fixture scan reports, so one table covers all the cases. The end-to-end test now covers a resolvable control key, bucket(4, missing), f(g(missing)), [id, missing], [missing, bucket(4, missing)], and s.missing. For each unresolvable case it checks that the warning names exactly the unresolvable columns, that the whole partitioning is dropped, and that a shuffle is planned. For [id, missing], only missing is listed.
| val partitioning = sequenceToOption( | ||
| kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog)) | ||
| .toImmutableArraySeq) | ||
| if (partitioning.isEmpty) { |
There was a problem hiding this comment.
Yes, I changed it that way. The rule now checks the key references recursively against relation first. If any can't be resolved, it logs the warning and uses None without calling toCatalystOpt; otherwise the existing path runs unchanged. The reported ordering uses the same check.
| logWarning( | ||
| log"Spark ignores the reported ${MDC(CLASS_NAME, kgp.getClass.getSimpleName)} " + | ||
| log"because the partition key columns cannot be resolved: " + | ||
| log"${MDC(COLUMN_NAMES, unresolvedRefs.map(_.describe()).mkString(", "))}. " + |
There was a problem hiding this comment.
Done, the column names are de-duplicated, and the [missing, bucket(4, missing)] case checks that missing is listed once. Agreed on the repeated logging; I left that as is, like the existing warning.
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing the second round, @anuragmantri. The up-front check in V2ScanPartitioningAndOrdering makes the fallback decision and the warning come from the same place, and dropping the whole partitioning (instead of keeping only the resolvable keys) is the safe choice for storage-partitioned join. I left a few more comments.
- Scope: Since the rule checks every key reference before calling
toCatalystOpt, thefailOnUnresolvedrefactoring inV2ExpressionUtilsis no longer needed for this fix. - Test coverage: The control case does not assert that SPJ is applied.
- Diagnosability: The new warnings do not name the relation.
- PR title and description: Could you update them to the final design?
- The title mentions only the partitioning, but the reported ordering is also dropped now.
- The first paragraph describes
toCatalystOpt/toCatalystTransformOptas the fix, but the decision is now made by the up-front check inV2ScanPartitioningAndOrdering. (SPARK-59248)looks inaccurate. TheNonefallback aftertoCatalystOptlong predates SPARK-59248, and SPARK-59248 is about keys pruned from the scan output, which are still resolvable against the relation. Before this PR, an unresolvable key failed the query.
| /** | ||
| * Shared implementation of `toCatalyst`/`toCatalystOpt`, keyed on `failOnUnresolved`. | ||
| */ | ||
| private def toCatalystCore( |
There was a problem hiding this comment.
Now that V2ScanPartitioningAndOrdering checks every key reference against relation before calling toCatalystOpt, the None-on-unresolved path here is reachable only from tests: partitioning() is the only production caller of toCatalystOpt, and toCatalystTransformOpt has no production caller. The only production-visible effect left is that a nested transform whose function cannot be loaded drops the partitioning instead of failing the query.
Could we keep V2ExpressionUtils unchanged except for resolveRefOpt? Then this PR becomes smaller, the contract of toCatalystOpt/toCatalystTransformOpt and the nested-transform behavior stay as they are, and the new KeyGroupedPartitioningSuite cases should still pass. If you prefer to keep it as a defensive change, could you mention it in the PR description?
There was a problem hiding this comment.
Agreed, I reverted it. V2ExpressionUtils is now unchanged except for adding resolveRefOpt, with resolveRef as a thin wrapper around it. I removed the V2ExpressionUtilsSuite tests that were tied to the refactor and kept tests for resolveRefOpt.
| case None => | ||
| assert(warnings.isEmpty, warnings) | ||
| assert(reportingScan.keyGroupedPartitioning.isDefined, keysString) |
There was a problem hiding this comment.
Could you also assert collectShuffles(plan).isEmpty for this control case? Otherwise, collectShuffles(plan).nonEmpty in the fallback cases would pass even if this fixture could never plan a storage-partitioned join.
BTW, CustomReportingCatalog extends InMemoryTableCatalog, which is not a FunctionCatalog, so the bucket(4, missing) and f(g(missing)) keys would be dropped even if their columns were resolvable. It's fine because the warning assertion distinguishes them, but a short comment would help.
There was a problem hiding this comment.
Done. The control case now asserts collectShuffles(plan).isEmpty, which confirms the fixture does plan a storage-partitioned join when the key resolves. I also added a comment that CustomReportingCatalog is not a FunctionCatalog, and that the warning assertion is what shows the unresolvable column causes the drop.
| .toImmutableArraySeq) | ||
| val unresolvedColumns = unresolvableColumns(kgp.keys().toImmutableArraySeq, relation) | ||
| val partitioning = if (unresolvedColumns.nonEmpty) { | ||
| logWarning( |
There was a problem hiding this comment.
This warning does not say which relation reports the key, and kgp.getClass.getSimpleName is KeyGroupedPartitioning in practice, so it's hard to find the connector at fault in a query with several V2 scans. Could you include the relation name like V2ScanRelationPushDown does, e.g., ${MDC(RELATION_NAME, relation.name)}, here and in the ordering warning below? In addition, the reason of the caught AnalysisException is dropped, so an ambiguous column is reported as "cannot be resolved" in the same way as a missing one.
There was a problem hiding this comment.
Done. Both warnings now include ${MDC(RELATION_NAME, relation.name)} and I replaced kgp.getClass.getSimpleName with the scan class, so the connector at fault is easy to find. When resolveRefOpt throws, the column is listed with the error condition, e.g. s.missing (FIELD_NOT_FOUND) or a (AMBIGUOUS_REFERENCE). The end-to-end tests check the relation name and the s.missing (FIELD_NOT_FOUND) entry.
| val partitioning = if (unresolvedColumns.nonEmpty) { | ||
| logWarning( | ||
| log"Spark ignores the reported ${MDC(CLASS_NAME, kgp.getClass.getSimpleName)} " + | ||
| log"because the partition key columns cannot be resolved: " + |
There was a problem hiding this comment.
FYI, the motivating case in #55518 (a relation narrowed for column-level UPDATE) is not a connector bug, so this warning would be logged whenever the narrowed relation omits a partition source column if #55518 lands as is. WARN looks fine for this PR because it helps to find connector bugs, but we may need to revisit the level there.
There was a problem hiding this comment.
Thanks for the heads-up. Agreed that WARN fits here, since an unresolvable key usually means a connector bug. I'll revisit the level in #55518, where a narrowed relation can legitimately omit a partition source column.
| V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation, relation.funCatalog) | ||
| d.copy(ordering = Some(ordering)) | ||
| val unresolvedColumns = | ||
| unresolvableColumns(scan.outputOrdering().toImmutableArraySeq, relation) |
There was a problem hiding this comment.
nit. scan.outputOrdering() is now called twice, and kgp.keys() above too. Could you keep the result in a local variable? A connector may build a new array on each call.
There was a problem hiding this comment.
Done, kgp.keys() and scan.outputOrdering() are each read once into a local variable now.
| logWarning( | ||
| log"Spark ignores the reported ordering because the ordering columns cannot be " + | ||
| log"resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(", "))}.") | ||
| d.copy(ordering = None) |
There was a problem hiding this comment.
Before this PR, a SupportsReportOrdering scan always had ordering = Some(...). With None here, DataSourceV2ScanExecBase.outputOrdering falls back to the partition-key-derived ordering when the scan keeps a key-grouped partitioning and spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled is true (the default), while a scan reporting an empty ordering (Some(Nil)) does not get it. It looks correct to me because the derived ordering does not depend on the dropped report. Could you add a short comment here and assert reportingScan.outputOrdering in the new ordering test so that this behavior is explicit?
There was a problem hiding this comment.
Agreed. I added a comment that with no reported ordering, DataSourceV2ScanExecBase.outputOrdering may still derive one from a kept key-grouped partitioning, and that this doesn't depend on the dropped report. I also updated the outputOrdering scaladoc, which said a reported ordering is always returned as-is. The ordering test now sets partitionKeyOrdering.enabled explicitly, asserts that reportingScan.outputOrdering is the derived id ASC, and has a control case where a resolvable reported ordering is kept without a warning.
| private def unresolvableColumns( | ||
| exprs: Seq[V2Expression], | ||
| relation: LogicalPlan): Seq[String] = { | ||
| def references(expr: V2Expression): Seq[NamedReference] = expr match { |
There was a problem hiding this comment.
nit. Could you add a short comment on why this does not use Expression.references()? ApplyTransform and BucketTransform override it with only their top-level arguments, so it would miss f(g(missing)).
There was a problem hiding this comment.
Done. I added a comment that ApplyTransform and BucketTransform override references() to return only their top-level arguments, so it would miss f(g(missing)).
| package org.apache.spark.sql.connector | ||
|
|
||
| import java.sql.Timestamp | ||
| import java.util |
There was a problem hiding this comment.
nit. This file uses java.util.Map fully qualified elsewhere (e.g., emptyProps). Could you use java.util.Map in CustomReportingCatalog and remove this import?
There was a problem hiding this comment.
Done, the fixture now uses java.util.Map and the import is removed.
There was a problem hiding this comment.
Thanks for the PR, @anuragmantri!
Splitting toCatalyst/toCatalystOpt onto a shared core keyed on failOnUnresolved is the right shape, and it keeps the blast radius where it belongs: I traced the path I was most worried about, and DistributionAndOrderingUtils still reaches the throwing toCatalyst for a write's ClusteredDistribution clustering and toCatalystOrdering for its required ordering, so nothing on the write path degrades silently. Two non-blocking items, both about the contract the *Opt variants now carry rather than about what they do.
One measurement to add to the "degradation is still partial" point at r4077137407, since the description states it but nothing pins it: with a reported key of id + 1 (a GeneralScalarExpression over a resolvable column), the pre-check finds no unresolvable column and toCatalystOpt throws from case _ =>, so the query still fails with id + 1 is not currently supported. Ran end to end against the new CustomReportingCatalog fixture on this head.
Non-blocking
- 1. The public
*Optvariants still throw, and say nothing about it: the scaladoc precision from r4117400462 landed onresolveRefOpt, which isprivate[sql].toCatalystOptandtoCatalystTransformOptare the public entry points and carry no scaladoc at all. inline:V2ExpressionUtils.scala:139 - 2. The ambiguous-reference arm of the rule's
catchis untested: the nested-field arm is covered end to end, ambiguity is not, in either suite, although theresolveRefOptscaladoc and the PR description both name it. inline:V2ScanPartitioningAndOrdering.scala:116
| query: LogicalPlan, | ||
| funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] = { | ||
| funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] = | ||
| toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = false) |
There was a problem hiding this comment.
Finding 1. The precision fix from r4117400462 landed on resolveRefOpt, which is private[sql]. toCatalystOpt and toCatalystTransformOpt are the public entry points, and they carry no scaladoc, so the split contract is only discoverable three levels down.
Measured on this head:
toCatalystOpt(FieldReference("missing"), plan) -> None
toCatalystOpt(FieldReference("s.missing"), plan) -> throws [FIELD_NOT_FOUND]
resolveRefOpt(FieldReference("a"), relationWithTwoAs) -> throws [AMBIGUOUS_REFERENCE]
toCatalystOpt(GeneralScalarExpression("+", [id, 1]), plan) -> throws "id + 1 is not currently supported"
This is not hypothetical. The rule in this PR had to wrap resolveRefOpt in try { ... } catch { case _: AnalysisException => true } for exactly that reason, and the next caller will read toCatalystOpt the way its name invites. A couple of lines each:
/**
* Variant of `toCatalyst` that returns `None` when a reference cannot be resolved. It still
* throws for a missing nested field or an ambiguous reference (see `resolveRefOpt`), and for an
* expression type this conversion does not handle.
*/
def toCatalystOpt(and the same on toCatalystTransformOpt.
There was a problem hiding this comment.
Following Dongjoon's suggestion, I reverted the V2ExpressionUtils changes, so toCatalystOpt and toCatalystTransformOpt keep their original contract and this PR no longer changes them.
| V2ExpressionUtils.resolveRefOpt[NamedExpression](ref, relation).isEmpty | ||
| } catch { | ||
| // A missing nested field or an ambiguous name throws instead of returning None. | ||
| case _: AnalysisException => true |
There was a problem hiding this comment.
Finding 2. This catch absorbs two distinct throws from resolveRefOpt, and only one of them is exercised. The end-to-end test covers s.missing (FIELD_NOT_FOUND); the ambiguous-reference arm has no test in this suite or in V2ExpressionUtilsSuite, although the resolveRefOpt scaladoc and the PR description both name it.
Building a v2 relation with two same-named columns to reach it end to end looks like more trouble than it is worth, so the cheap place is V2ExpressionUtilsSuite, pinning the precondition this catch rests on:
test("SPARK-59721: resolveRefOpt throws for an ambiguous reference") {
val dup = LocalRelation.apply(
AttributeReference("a", StringType)(), AttributeReference("a", StringType)())
intercept[AnalysisException] {
V2ExpressionUtils.resolveRefOpt[NamedExpression](FieldReference("a"), dup)
}
}I ran that relation on this head and it throws [AMBIGUOUS_REFERENCE] Reference `a` is ambiguous, could be: [`a`, `a`]. checkError with condition = "AMBIGUOUS_REFERENCE" would be the better assertion; I did not run it, so the parameter map would need checking.
There was a problem hiding this comment.
Added your test to V2ExpressionUtilsSuite, using checkError with AMBIGUOUS_REFERENCE, together with the FIELD_NOT_FOUND case for s.missing. Both are the preconditions this catch relies on.
anuragmantri
left a comment
There was a problem hiding this comment.
Thanks for the third pass, @dongjoon-hyun. I addressed all comments in the latest commit. Details are in the inline replies.
- Scope: Reverted the
failOnUnresolvedrefactoring.V2ExpressionUtilsnow only addsresolveRefOpt, so thetoCatalyst*contracts are unchanged. - Test coverage: The control case now asserts that storage-partitioned join is applied, and the ordering test has a control case where a resolvable reported ordering is kept.
- Diagnosability: Both warnings name the relation and the scan class, and a caught
AnalysisExceptionis reported with its error condition. - PR title and description: I updated the title and rewrote the description to match the code.
@peter-toth thanks for the review and for tracing the write path. I addressed both items in the latest commit.
- Measurement: Added an end-to-end test that pins it: a reported partition key of
id + 1, or a nested transform whose function can't be loaded likef(g(id)), still fails the query with_LEGACY_ERROR_TEMP_3054. The PR description lists both as unchanged. - 1.
*Optscaladoc: Following Dongjoon's suggestion, I reverted theV2ExpressionUtilschanges, sotoCatalystOptandtoCatalystTransformOptkeep their master contract and this PR no longer changes them. - 2. Ambiguous-reference: Added your test to
V2ExpressionUtilsSuite, usingcheckErrorwithAMBIGUOUS_REFERENCE, together with theFIELD_NOT_FOUNDcase fors.missing. Both are the preconditions the rule'scatchrelies on.
| /** | ||
| * Shared implementation of `toCatalyst`/`toCatalystOpt`, keyed on `failOnUnresolved`. | ||
| */ | ||
| private def toCatalystCore( |
There was a problem hiding this comment.
Agreed, I reverted it. V2ExpressionUtils is now unchanged except for adding resolveRefOpt, with resolveRef as a thin wrapper around it. I removed the V2ExpressionUtilsSuite tests that were tied to the refactor and kept tests for resolveRefOpt.
| case None => | ||
| assert(warnings.isEmpty, warnings) | ||
| assert(reportingScan.keyGroupedPartitioning.isDefined, keysString) |
There was a problem hiding this comment.
Done. The control case now asserts collectShuffles(plan).isEmpty, which confirms the fixture does plan a storage-partitioned join when the key resolves. I also added a comment that CustomReportingCatalog is not a FunctionCatalog, and that the warning assertion is what shows the unresolvable column causes the drop.
| .toImmutableArraySeq) | ||
| val unresolvedColumns = unresolvableColumns(kgp.keys().toImmutableArraySeq, relation) | ||
| val partitioning = if (unresolvedColumns.nonEmpty) { | ||
| logWarning( |
There was a problem hiding this comment.
Done. Both warnings now include ${MDC(RELATION_NAME, relation.name)} and I replaced kgp.getClass.getSimpleName with the scan class, so the connector at fault is easy to find. When resolveRefOpt throws, the column is listed with the error condition, e.g. s.missing (FIELD_NOT_FOUND) or a (AMBIGUOUS_REFERENCE). The end-to-end tests check the relation name and the s.missing (FIELD_NOT_FOUND) entry.
| val partitioning = if (unresolvedColumns.nonEmpty) { | ||
| logWarning( | ||
| log"Spark ignores the reported ${MDC(CLASS_NAME, kgp.getClass.getSimpleName)} " + | ||
| log"because the partition key columns cannot be resolved: " + |
There was a problem hiding this comment.
Thanks for the heads-up. Agreed that WARN fits here, since an unresolvable key usually means a connector bug. I'll revisit the level in #55518, where a narrowed relation can legitimately omit a partition source column.
| V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation, relation.funCatalog) | ||
| d.copy(ordering = Some(ordering)) | ||
| val unresolvedColumns = | ||
| unresolvableColumns(scan.outputOrdering().toImmutableArraySeq, relation) |
There was a problem hiding this comment.
Done, kgp.keys() and scan.outputOrdering() are each read once into a local variable now.
| logWarning( | ||
| log"Spark ignores the reported ordering because the ordering columns cannot be " + | ||
| log"resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(", "))}.") | ||
| d.copy(ordering = None) |
There was a problem hiding this comment.
Agreed. I added a comment that with no reported ordering, DataSourceV2ScanExecBase.outputOrdering may still derive one from a kept key-grouped partitioning, and that this doesn't depend on the dropped report. I also updated the outputOrdering scaladoc, which said a reported ordering is always returned as-is. The ordering test now sets partitionKeyOrdering.enabled explicitly, asserts that reportingScan.outputOrdering is the derived id ASC, and has a control case where a resolvable reported ordering is kept without a warning.
| private def unresolvableColumns( | ||
| exprs: Seq[V2Expression], | ||
| relation: LogicalPlan): Seq[String] = { | ||
| def references(expr: V2Expression): Seq[NamedReference] = expr match { |
There was a problem hiding this comment.
Done. I added a comment that ApplyTransform and BucketTransform override references() to return only their top-level arguments, so it would miss f(g(missing)).
| package org.apache.spark.sql.connector | ||
|
|
||
| import java.sql.Timestamp | ||
| import java.util |
There was a problem hiding this comment.
Done, the fixture now uses java.util.Map and the import is removed.
| query: LogicalPlan, | ||
| funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] = { | ||
| funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] = | ||
| toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = false) |
There was a problem hiding this comment.
Following Dongjoon's suggestion, I reverted the V2ExpressionUtils changes, so toCatalystOpt and toCatalystTransformOpt keep their original contract and this PR no longer changes them.
| V2ExpressionUtils.resolveRefOpt[NamedExpression](ref, relation).isEmpty | ||
| } catch { | ||
| // A missing nested field or an ambiguous name throws instead of returning None. | ||
| case _: AnalysisException => true |
There was a problem hiding this comment.
Added your test to V2ExpressionUtilsSuite, using checkError with AMBIGUOUS_REFERENCE, together with the FIELD_NOT_FOUND case for s.missing. Both are the preconditions this catch relies on.
There was a problem hiding this comment.
Re-checked through d06b704. Findings 1 and 2 are resolved. The failOnUnresolved refactor is reverted, so toCatalystOpt keeps master's contract. V2ExpressionUtilsSuite now pins both AMBIGUOUS_REFERENCE and FIELD_NOT_FOUND. Nothing regressed, and the code change reads correct to me. Only text is left.
Minor
- 3. A rewritten test comment says an empty partitioning report yields
Some(Nil)(new): the partitioning pass turns an empty report intoNone, so the parenthetical holds for the ordering pass only. inline:MergeSubplansSuite.scala:2933 - 4. SPARK-59721 still describes the first design (new): its summary covers only the partitioning, and its description still names
toCatalystOptreturningNoneas the fix, plus the SPARK-59248 sentence the PR description dropped. - 5. A test bullet in the description is cut off (new): the second
V2ExpressionUtilsSuitebullet ends at "for an ambiguous", and the description becomes the commit message.
| // A source implementing SupportsReportPartitioning/SupportsReportOrdering but reporting nothing | ||
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unconditionally). An | ||
| // empty report carries no partitioning/ordering to drop, so the merge should still proceed -- | ||
| // the gate tests the inner Seq, not the Option. The fused plan is the plain column union. | ||
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unless a reported | ||
| // column cannot be resolved). An empty report carries no partitioning/ordering to drop, so the | ||
| // merge should still proceed -- the gate tests the inner Seq, not the Option. The fused plan is | ||
| // the plain column union. |
There was a problem hiding this comment.
Finding 3. The rewritten parenthetical is accurate for the ordering pass only. The partitioning pass never yields Some(Nil):
- an
UnknownPartitioningreport givesNone(V2ScanPartitioningAndOrdering.scala:78); - a
KeyGroupedPartitioningwith no keys givesNonetoo, becauseSome(Nil).filter(_.exists(...))isNone(V2ScanPartitioningAndOrdering.scala:77, whose own comment says "likewise when the source reported no key at all").
That has been the case since SPARK-59248, so the keyGroupedPartitioning = Some(Nil) half of this test is a state the rule does not produce. Something like:
| // A source implementing SupportsReportPartitioning/SupportsReportOrdering but reporting nothing | |
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unconditionally). An | |
| // empty report carries no partitioning/ordering to drop, so the merge should still proceed -- | |
| // the gate tests the inner Seq, not the Option. The fused plan is the plain column union. | |
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unless a reported | |
| // column cannot be resolved). An empty report carries no partitioning/ordering to drop, so the | |
| // merge should still proceed -- the gate tests the inner Seq, not the Option. The fused plan is | |
| // the plain column union. | |
| // A source implementing SupportsReportOrdering but reporting no ordering yields Some(Nil), not | |
| // None (V2ScanPartitioningAndOrdering sets the ordering unless a reported column cannot be | |
| // resolved). The partitioning pass turns an empty report into None instead, so the rule never | |
| // produces the keyGroupedPartitioning = Some(Nil) case below. An empty report carries no | |
| // partitioning/ordering to drop, so the merge should still proceed -- the gate tests the inner | |
| // Seq, not the Option. The fused plan is the plain column union. |
There was a problem hiding this comment.
Thanks, applied your wording. The comment now says only the ordering pass yields Some(Nil), and that the partitioning pass turns an empty report into None.
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing the third round, @anuragmantri. V2ExpressionUtils only adds resolveRefOpt now, and both warnings name the relation and the scan class. I left a few more comments. Most of them are minor.
- Warning message: The unresolvable columns are rendered via
describe(), so a connector-definedNamedReferencecan be printed asClass@hash. In addition, the last sentence of the partitioning warning covers only storage-partitioned join. - Test coverage: The new tests use only the built-in expression classes, and a report which mixes an unconvertible key with an unresolvable column is not pinned.
- Comments and docs: The sibling comment in
PlanMerger, the doc ofpartitionKeyOrdering.enabled, and the comment on the reference walker. - Pre-existing behavior: I left two FYI notes on the ordering pass. They are not for this PR.
| * Variant of `resolveRef` that returns `None` if no attribute matches the reference. It still | ||
| * throws for a missing nested field or an ambiguous reference. | ||
| */ | ||
| private[sql] def resolveRefOpt[T <: NamedExpression]( |
There was a problem hiding this comment.
nit. The type parameter looks unnecessary now. All callers other than resolveRef pass [NamedExpression], and none of them uses the typed result. How about making this non-generic and keeping the cast in resolveRef like before?
private[sql] def resolveRefOpt(
ref: NamedReference, plan: LogicalPlan): Option[NamedExpression] = {
plan.resolve(ref.fieldNames.toImmutableArraySeq, conf.resolver)
}
def resolveRef[T <: NamedExpression](ref: NamedReference, plan: LogicalPlan): T = {
resolveRefOpt(ref, plan).getOrElse {
...
}.asInstanceOf[T]
}Then, the explicit [NamedExpression] at the call sites and the NamedExpression import in V2ScanPartitioningAndOrdering can be removed.
There was a problem hiding this comment.
Done. resolveRefOpt is non-generic and returns Option[NamedExpression], resolveRef keeps the cast, and I removed the type arguments at the call sites and the NamedExpression import in V2ScanPartitioningAndOrdering.
| /** | ||
| * Returns the output ordering for this scan. When the source reports ordering via | ||
| * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise, when the output | ||
| * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it, that ordering is |
There was a problem hiding this comment.
nit. The same condition is described in the doc of spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled in SQLConf and in docs/sql-performance-tuning.md, e.g., ... reports a KeyedPartitioning but does not report explicit ordering via SupportsReportOrdering. After this PR, a source whose reported ordering is ignored also gets the key-derived ordering, as the new test asserts. They are incomplete rather than wrong, so this is optional. If you update them, could you update both? For example, ..., or reports one that Spark ignores because it references a column that cannot be resolved.
There was a problem hiding this comment.
Done. I updated both the partitionKeyOrdering.enabled doc in SQLConf and docs/sql-performance-tuning.md with your wording
| * Returns the output ordering for this scan. When the source reports ordering via | ||
| * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise, when the output | ||
| * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it, that ordering is | ||
| * returned as-is. Otherwise, when the output |
There was a problem hiding this comment.
FYI, this is pre-existing and not for this PR. returned as-is also applies to a sort order on a column which is pruned out of the scan output, because the ordering pass resolves against the full relation. DataSourceV2ScanRelation.doCanonicalize handles it with takeWhile(_.references.subsetOf(outputSet)), but outputOrdering does not.
A row-based scan is safe because the ProjectExec above it truncates such a sort order. For a columnar scan without ProjectExec, it looks like the k-way merge of GroupPartitionsExec (new LazyCodeGenOrdering(kWayMergeOrdering, child.output)) can fail to bind it when spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled is true (false by default). This is from reading the code, and I have not reproduced it yet. I am checking it separately.
There was a problem hiding this comment.
Thanks, I reproduced it on both this branch and master. With a columnar source that reports KeyGroupedPartitioning(k) and ordering [k, a, c], prunes c, and has preserveOrderingOnCoalesce enabled, a window over (PARTITION BY k ORDER BY a) plans GroupPartitions(SortedMerge) -> ColumnarToRow -> BatchScan[k, a] with no ProjectExec in between. The task fails with INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND ("Could not find c#2 in [k#0,a#1]"). The row-based variant is fine because ProjectExec truncates the ordering.
Trimming DataSourceV2ScanExecBase.outputOrdering with takeWhile(_.references.subsetOf(outputSet)), as doCanonicalize does, fixes it and keeps the related suites green. Since it's pre-existing and independent of this PR, I'll keep it out of here and file a separate JIRA and PR with a test, unless you already have one in progress.
| log"${MDC(RELATION_NAME, relation.name)} (scan " + | ||
| log"${MDC(CLASS_NAME, scan.getClass.getName)}) because the partition key columns " + | ||
| log"cannot be resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(", "))}. " + | ||
| log"Storage-partitioned join will not be applied for this scan.") |
There was a problem hiding this comment.
nit. This sentence is narrower than what happens. Without the reported partitioning, storage-partitioned aggregates and windows and the partition-key-derived ordering are lost too (cf. storage-partitioned operations (joins, aggregates, and windows) in SQLConf). In addition, when spark.sql.sources.v2.bucketing.shuffle.enabled is true and the other side is keyed, a storage-partitioned join can still be planned by shuffling only this side. Could you remove this sentence like the ordering warning below, or generalize it?
There was a problem hiding this comment.
Removed the sentence, so it matches the ordering warning.
| @@ -75,8 +89,43 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging { | |||
| private def ordering(plan: LogicalPlan) = plan.transformDownWithPruning( | |||
| _.containsPattern(DATA_SOURCE_V2_SCAN_RELATION)) { | |||
| case d @ ExtractV2ScanInfo(relation, scan: SupportsReportOrdering, _) => | |||
There was a problem hiding this comment.
FYI, this is also pre-existing and not for this PR. Unlike the partitioning pass (if d.keyGroupedPartitioning.isEmpty), the ordering pass has no guard and recomputes the ordering on every application. For a resolvable nested sort key like s.x, plan.resolve returns a new Alias(GetStructField(...)) with a fresh ExprId each time, so the rule is not idempotent for such a report.
It looks like the Once batch idempotence check of RuleExecutor would fail in testing mode. In production, the reported SortOrder(Alias(...)) would not match a required ordering because the Alias is not stripped. No existing test reports a nested sort key, but this PR adds the struct column s and reportedOrdering to the fixture, so it is easy to hit by adding such a case. This is from reading the code, not from a run.
There was a problem hiding this comment.
Thanks, I reproduced it. With a reported ordering on the nested field s.x, the "Early Filter and Projection Push-Down" batch fails its idempotence check in testing, because each run resolves s.x to a new Alias(GetStructField(...)). With that check skipped, the aliased sort order never satisfies a required ordering on s.x, so Spark adds a redundant SortExec. Trimming aliases from the converted ordering fixes both, and the existing suites still pass.
Since it's pre-existing and changes planning behavior on its own, I'll keep it out of this PR and file a separate JIRA and PR. I also left out a nested sort key test here for the same reason.
There was a problem hiding this comment.
I see you've fixed it in SPARK-59899. I rebased this PR on top of it and merged the outputOrdering scaladoc with your new wording; the new SPARK-59899 test passes together with the tests here.
| Column.create("data", StringType), | ||
| Column.create("s", new StructType().add("x", IntegerType))) | ||
| createTable(table, reportingColumns, Array(identity("id")), catalog = reportingCatalog) | ||
| sql(s"INSERT INTO $customReportingCatalogName.ns.$table VALUES " + |
There was a problem hiding this comment.
nit. customReportingTableName is defined above, but this line and the SELECT in customReportingJoinDf assemble the same string again. Could you use the constant?
There was a problem hiding this comment.
Done, the INSERT and the SELECT in customReportingJoinDf both use customReportingTableName now.
| ApplyTransform("f", Seq(ApplyTransform("g", Seq(FieldReference("missing"))))) | ||
| // CustomReportingCatalog is not a FunctionCatalog, so a `bucket` key would be dropped even with | ||
| // a resolvable column. The warning assertion shows the unresolvable column is what drops it. | ||
| val cases = Seq[(Seq[Expression], Option[String])]( |
There was a problem hiding this comment.
All the new cases use the built-in expression classes (FieldReference, ApplyTransform, BucketTransform, SortValue), so the unresolvable path is not covered for connector-defined Transform/SortOrder/NamedReference classes. That is where walking children() matters most. For example, MyIdentityTransform.references() in DataSourceV2Suite returns an empty array.
Could you add one case which reads the existing OrderAndPartitionAwareDataSource (and its Java twin, JavaOrderAndPartitionAwareDataSource) with partitionKeys=i and orderKeys=missing, and verify that the query succeeds, the reported ordering is dropped, and the warning names missing? It also covers my describe() comment in V2ScanPartitioningAndOrdering.
There was a problem hiding this comment.
Added SPARK-59721: ignore an unresolvable ordering reported with connector-defined classes to DataSourceV2Suite, next to the existing ordering and partitioning reporting test. It reads both the Scala and Java OrderAndPartitionAwareDataSource with partitionKeys=i and orderKeys=missing, and checks that the query succeeds, the reported ordering is dropped, and the warning names missing.
| } | ||
| } | ||
|
|
||
| test("SPARK-59721: a reported partition key that cannot be converted still fails") { |
There was a problem hiding this comment.
This pins id + 1 and f(g(id)), where every column resolves. Note that the pre-check runs over the whole report before the conversion, so a report which mixes an unconvertible key with an unresolvable column (e.g., [id + 1, identity(missing)], or a single key missing + 1) does not fail anymore. It succeeds with a warning which mentions only missing, and it starts to fail with _LEGACY_ERROR_TEMP_3054 once the column becomes resolvable.
It is consistent with the PR description (Keys whose columns all resolve take the existing conversion path unchanged), so no code change is needed. Could you add one mixed case to the fallback test above, e.g., (Seq(<id + 1>, identity("missing")), Some("missing")), so that this boundary is explicit?
There was a problem hiding this comment.
Added [id + 1, identity(missing)] to the fallback test. It succeeds with a warning that names only missing.
| * of the table's own, so a test can report keys on columns the table does not have. | ||
| */ | ||
| class CustomReportingCatalog extends InMemoryTableCatalog { | ||
| @volatile var reportedKeys: Seq[Expression] = Seq.empty |
There was a problem hiding this comment.
nit. @volatile looks unnecessary. These fields are written by the test thread and read only in build(), which the optimizer invokes on the same thread. The other in-memory test catalogs use plain var for such knobs.
There was a problem hiding this comment.
Done, removed @volatile.
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unconditionally). An | ||
| // empty report carries no partitioning/ordering to drop, so the merge should still proceed -- | ||
| // the gate tests the inner Seq, not the Option. The fused plan is the plain column union. | ||
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unless a reported |
There was a problem hiding this comment.
nit. Since this comment is updated, could you also update the sibling comment above mergeDegradesReporting in PlanMerger.scala? It says partitioning pass keeps the report whenever any key survives, which is no longer literally true. For example, [identity(id), identity(missing)] drops the whole report although id survives in the scan output (the new test pins it). The merge logic and the conclusion of that comment still hold because the new check depends only on the reported columns and relation, so adding unless a reported column cannot be resolved there would be enough.
There was a problem hiding this comment.
Done, I added "unless a reported column cannot be resolved" to the comment above mergeDegradesReporting.
anuragmantri
left a comment
There was a problem hiding this comment.
Thank you @dongjoon-hyun and @peter-toth. I addressed all your comments.
| * Variant of `resolveRef` that returns `None` if no attribute matches the reference. It still | ||
| * throws for a missing nested field or an ambiguous reference. | ||
| */ | ||
| private[sql] def resolveRefOpt[T <: NamedExpression]( |
There was a problem hiding this comment.
Done. resolveRefOpt is non-generic and returns Option[NamedExpression], resolveRef keeps the cast, and I removed the type arguments at the call sites and the NamedExpression import in V2ScanPartitioningAndOrdering.
| /** | ||
| * Returns the output ordering for this scan. When the source reports ordering via | ||
| * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise, when the output | ||
| * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it, that ordering is |
There was a problem hiding this comment.
Done. I updated both the partitionKeyOrdering.enabled doc in SQLConf and docs/sql-performance-tuning.md with your wording
| log"${MDC(RELATION_NAME, relation.name)} (scan " + | ||
| log"${MDC(CLASS_NAME, scan.getClass.getName)}) because the partition key columns " + | ||
| log"cannot be resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(", "))}. " + | ||
| log"Storage-partitioned join will not be applied for this scan.") |
There was a problem hiding this comment.
Removed the sentence, so it matches the ordering warning.
| // This is not `V2Expression.references()` because `ApplyTransform` and `BucketTransform` | ||
| // override it to return only their top-level arguments, which would miss `f(g(missing))`. |
There was a problem hiding this comment.
Updated the comment to your wording. I also widened the catch comment and the resolveRefOpt scaladoc to "a nested-field extraction error or an ambiguous reference", which also covers AMBIGUOUS_REFERENCE_TO_FIELDS and INVALID_EXTRACT_BASE_FIELD_TYPE.
| if (V2ExpressionUtils.resolveRefOpt[NamedExpression](ref, relation).isDefined) { | ||
| None | ||
| } else { | ||
| Some(ref.describe()) |
There was a problem hiding this comment.
Thanks for the catch. The name is now built from fieldNames like resolveRef does, and de-duplicated on the string, so a connector-defined NamedReference prints its column name. The new DataSourceV2Suite test covers it. I checked that it fails when describe() is used.
| * of the table's own, so a test can report keys on columns the table does not have. | ||
| */ | ||
| class CustomReportingCatalog extends InMemoryTableCatalog { | ||
| @volatile var reportedKeys: Seq[Expression] = Seq.empty |
There was a problem hiding this comment.
Done, removed @volatile.
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unconditionally). An | ||
| // empty report carries no partitioning/ordering to drop, so the merge should still proceed -- | ||
| // the gate tests the inner Seq, not the Option. The fused plan is the plain column union. | ||
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unless a reported |
There was a problem hiding this comment.
Done, I added "unless a reported column cannot be resolved" to the comment above mergeDegradesReporting.
| // A source implementing SupportsReportPartitioning/SupportsReportOrdering but reporting nothing | ||
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unconditionally). An | ||
| // empty report carries no partitioning/ordering to drop, so the merge should still proceed -- | ||
| // the gate tests the inner Seq, not the Option. The fused plan is the plain column union. | ||
| // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unless a reported | ||
| // column cannot be resolved). An empty report carries no partitioning/ordering to drop, so the | ||
| // merge should still proceed -- the gate tests the inner Seq, not the Option. The fused plan is | ||
| // the plain column union. |
There was a problem hiding this comment.
Thanks, applied your wording. The comment now says only the ordering pass yields Some(Nil), and that the partitioning pass turns an empty report into None.
| * Returns the output ordering for this scan. When the source reports ordering via | ||
| * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise, when the output | ||
| * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it, that ordering is | ||
| * returned as-is. Otherwise, when the output |
There was a problem hiding this comment.
Thanks, I reproduced it on both this branch and master. With a columnar source that reports KeyGroupedPartitioning(k) and ordering [k, a, c], prunes c, and has preserveOrderingOnCoalesce enabled, a window over (PARTITION BY k ORDER BY a) plans GroupPartitions(SortedMerge) -> ColumnarToRow -> BatchScan[k, a] with no ProjectExec in between. The task fails with INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND ("Could not find c#2 in [k#0,a#1]"). The row-based variant is fine because ProjectExec truncates the ordering.
Trimming DataSourceV2ScanExecBase.outputOrdering with takeWhile(_.references.subsetOf(outputSet)), as doCanonicalize does, fixes it and keeps the related suites green. Since it's pre-existing and independent of this PR, I'll keep it out of here and file a separate JIRA and PR with a test, unless you already have one in progress.
| @@ -75,8 +89,43 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging { | |||
| private def ordering(plan: LogicalPlan) = plan.transformDownWithPruning( | |||
| _.containsPattern(DATA_SOURCE_V2_SCAN_RELATION)) { | |||
| case d @ ExtractV2ScanInfo(relation, scan: SupportsReportOrdering, _) => | |||
There was a problem hiding this comment.
Thanks, I reproduced it. With a reported ordering on the nested field s.x, the "Early Filter and Projection Push-Down" batch fails its idempotence check in testing, because each run resolves s.x to a new Alias(GetStructField(...)). With that check skipped, the aliased sort order never satisfies a required ordering on s.x, so Spark adds a redundant SortExec. Trimming aliases from the converted ordering fixes both, and the existing suites still pass.
Since it's pre-existing and changes planning behavior on its own, I'll keep it out of this PR and file a separate JIRA and PR. I also left out a nested sort key test here for the same reason.
|
To @anuragmantri , please resolve the conflicts, too. |
e3e46a2 to
9eec6fc
Compare
peter-toth
left a comment
There was a problem hiding this comment.
Re-checked through 9eec6fc. Findings 3 and 4 are resolved. The MergeSubplansSuite comment took the suggestion, and SPARK-59721's summary and description now match the design. Nothing regressed, and the code still reads correct to me. The three new tests for the fix fail against master's V2ScanPartitioningAndOrdering with Unable to resolve missing given [...] and pass on this head.
One pre-existing issue looks like it got dropped. The latest reply on the nested sort key thread (r4150116207) says SPARK-59899 fixed it. SPARK-59899 only trims sort orders on pruned columns in outputOrdering, so the ordering pass still resolves s.x to a fresh Alias on every run. On this head, a reported ordering on s.x through CustomReportingCatalog still fails with Once strategy's idempotence is broken for batch Early Filter and Projection Push-Down. So the separate JIRA planned at r4149827763 is still needed.
Minor
- 5. A test bullet in the description is still cut off (round 2): the second
V2ExpressionUtilsSuitebullet ends at "for an ambiguous", and the description becomes the commit message. - 6. The testing section misses two of the new tests (new): it lists neither the
DataSourceV2Suitetest for connector-defined classes nor the mixed cases like[id + 1, identity(missing)].
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing the fourth round, @anuragmantri. I reviewed the latest commit, 9eec6fc, and left 9 inline comments. I did not find a regression, and none of the comments is a blocker. The new tests for the connector-defined classes and the mixed cases look good to me. Here is a summary, ordered by priority.
Correctness
V2ScanPartitioningAndOrderingL122: the pre-check walkschildren(), while the conversion resolves throughTransform.arguments()andSortOrder.expression(), so a connector that overrideschildren()can still fail withUnable to resolve ....V2ScanPartitioningAndOrderingL125: a reference with emptyfieldNames()still fails the query instead of being ignored. It is not a regression.
Docs and comments
V2ScanPartitioningAndOrderingL55: the connector API Javadoc (KeyGroupedPartitioning#keys(),SupportsReportOrdering#outputOrdering()) does not describe the new behavior.DataSourceV2ScanExecBaseL144: the partial reflow leaves a short line.
Tests
KeyGroupedPartitioningSuiteL3629: the warning assertions are duplicated across the two tests, and the message prefix is not asserted.KeyGroupedPartitioningSuiteL3608: the nested keyf(g(col))is built by hand twice.KeyGroupedPartitioningSuiteL3560: the catalog conf is set before thetry. This one is optional.
Cleanup
V2ScanPartitioningAndOrderingL120: the local helper has the same name asV2Expression.references().V2ScanPartitioningAndOrderingL54:keysis converted withtoImmutableArraySeqtwice.
Items 1 and 3 are the most important ones, since both are about how connectors see the new behavior. The others are minor.
| // `f(g(missing))`, and a connector-defined `Transform` can return anything. | ||
| def references(expr: V2Expression): Seq[NamedReference] = expr match { | ||
| case ref: NamedReference => Seq(ref) | ||
| case other => other.children().toImmutableArraySeq.flatMap(references) |
There was a problem hiding this comment.
This walk can still diverge from the conversion it guards. My earlier suggestion for the comment above was incomplete. The conversion does not use children(): toCatalystTransformOpt reads Transform.arguments() through the NamedTransform, IdentityTransform and BucketTransform extractors, and toCatalystOpt reads SortOrder.expression() through SortValue. The two agree only because of the default methods Transform.children() (arguments()) and SortOrder.children() ({expression()}). A connector can override them, which is the same reason the comment gives for not using references().
For example, a connector Transform named identity whose arguments() is [ref("missing")] but whose children() returns an empty array passes this check. Then IdentityTransform.unapply and resolveRef fail the query with Unable to resolve missing given [...], which is the failure this PR fixes. Conversely, a children() that returns extra references drops a report that would convert. No in-tree class overrides children(), so this is hardening.
Could you walk the same accessors as the conversion, and use children() only for the other expressions? For example, with SortOrder and Transform from org.apache.spark.sql.connector.expressions,
def references(expr: V2Expression): Seq[NamedReference] = expr match {
case ref: NamedReference => Seq(ref)
case t: Transform => t.arguments().toImmutableArraySeq.flatMap(references)
case s: SortOrder => references(s.expression())
case other => other.children().toImmutableArraySeq.flatMap(references)
}The existing cases like [id + 1, identity(missing)] and missing + 1 keep their behavior, because GeneralScalarExpression still goes through children(). The comment above would need the same update.
There was a problem hiding this comment.
Thanks, good catch. Done. The walk now reads Transform.arguments() and SortOrder.expression(), the same accessors the conversion uses, and falls back to children() only for other expressions such as GeneralScalarExpression. I updated the comment accordingly. I also added connector-defined Transform and SortOrder test classes whose children() disagrees in both directions: a missing reference hidden from children(), which used to fail with Unable to resolve missing, and an extra reference only in children(), which used to drop a valid partitioning.
| case other => other.children().toImmutableArraySeq.flatMap(references) | ||
| } | ||
| exprs.flatMap(references).flatMap { ref => | ||
| val name = ref.fieldNames.toImmutableArraySeq.quoted |
There was a problem hiding this comment.
Minor: A reference with empty fieldNames() still fails the query instead of being ignored. quoted here runs outside the try, and MultipartIdentifierHelper throws _LEGACY_ERROR_TEMP_1186 for empty parts. Moving it into the try is not enough either, because resolveRefOpt then reaches nameParts.head (expressions/package.scala:275) and throws NoSuchElementException, which the catch does not handle. This is not a regression, since the query failed with NoSuchElementException before this PR, and it needs a malformed connector reference. Could you treat an empty fieldNames as unresolvable before calling quoted and resolveRefOpt?
There was a problem hiding this comment.
Done. An empty fieldNames() is now treated as unresolvable before quoted or resolveRefOpt is called, and it is shown as <empty> in the warning. I added an ordering case for it, which used to fail with "Multi-part identifier cannot be empty".
| .toImmutableArraySeq) | ||
| val keys = kgp.keys() | ||
| val unresolvedColumns = unresolvableColumns(keys.toImmutableArraySeq, relation) | ||
| val partitioning = if (unresolvedColumns.nonEmpty) { |
There was a problem hiding this comment.
Could you document this behavior change in the connector API? Connector authors are the ones who need to know it. Before this PR, a reported key on a renamed or misspelled column failed the query in the connector's own tests. Now, storage-partitioned join or the reported ordering silently goes away, and the only signal is a WARN in the driver log. However, SupportsReportPartitioning#outputPartitioning(), KeyGroupedPartitioning#keys() and SupportsReportOrdering#outputOrdering() say nothing about what the references are resolved against, and the doc changes in this PR cover only the derived ordering. SupportsRuntimeV2Filtering#filterAttributes() already documents its requirement: "Each reference must resolve against the scan relation output when Spark builds it."
For example, in the Javadoc of KeyGroupedPartitioning#keys(), and similarly for SupportsReportOrdering#outputOrdering():
* <p>
* Spark resolves the column references in these expressions against the table columns,
* including metadata columns and columns pruned from the scan output. If any of them cannot
* be resolved, Spark ignores the whole reported partitioning and logs a warning.Keys whose columns resolve but cannot be converted, e.g. id + 1, still fail, so the sentence should cover only unresolvable references.
There was a problem hiding this comment.
Done. I added your sentence to the Javadoc of KeyGroupedPartitioning#keys() and SupportsReportOrdering#outputOrdering(), covering only unresolvable references. I also checked the "including metadata columns" part: a reported ordering on a metadata column resolves even when the query does not read it, since LogicalPlan.resolve falls back to the metadata output.
| * is a `KeyedPartitioning` and | ||
| * `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on, each partition | ||
| * contains rows where the key expressions evaluate to a single constant value, so the data | ||
| * is trivially sorted by those expressions within the partition. |
There was a problem hiding this comment.
Nit: The reflow stops here and leaves a short line. Could you reflow the rest of the paragraph?
| * is a `KeyedPartitioning` and | |
| * `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on, each partition | |
| * contains rows where the key expressions evaluate to a single constant value, so the data | |
| * is trivially sorted by those expressions within the partition. | |
| * is a `KeyedPartitioning` and `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` | |
| * is on, each partition contains rows where the key expressions evaluate to a single constant | |
| * value, so the data is trivially sorted by those expressions within the partition. |
| unresolvedColumns match { | ||
| case Some(columnsString) => | ||
| assert(warnings.nonEmpty, s"no warning for keys [$keysString]") | ||
| assert(warnings.forall(_.contains(s"reported by $customReportingTableName ")), warnings) |
There was a problem hiding this comment.
Minor: These fragment assertions are repeated in the ordering test below (L3660-L3662), and none of them pins the Spark ignores the KeyGroupedPartitioning reported by prefix. Since the MDC values are appended as they are, the whole message is deterministic. Could you build the expected message once and compare it as a whole? For example,
private def ignoredReportWarning(report: String, kind: String, columns: String): String =
s"Spark ignores the $report reported by $customReportingTableName " +
s"(scan ${classOf[CustomReportingScan].getName}) because the $kind columns " +
s"cannot be resolved: $columns."and
assert(warnings.toSet ==
Set(ignoredReportWarning("KeyGroupedPartitioning", "partition key", columnsString)),
keysString)toSet also absorbs the repeated warning from the idempotence check of the Once batch.
|
|
||
| test("SPARK-59721: an unresolvable reported partition key falls back to unknown partitioning") { | ||
| val nestedTransform = | ||
| ApplyTransform("f", Seq(ApplyTransform("g", Seq(FieldReference("missing"))))) |
There was a problem hiding this comment.
Nit: The same nested key is built again at L3676 with id. Could you add a helper next to plusOneKey? For example,
private def nestedKey(column: String): Expression =
ApplyTransform("f", Seq(ApplyTransform("g", Seq(FieldReference(column)))))Then both become one-line cases like the plusOneKey ones.
| new GeneralScalarExpression("+", Array[Expression](FieldReference(column), literal(1))) | ||
|
|
||
| private def withCustomReportingTable(f: CustomReportingCatalog => Unit): Unit = { | ||
| spark.conf.set(s"spark.sql.catalog.$customReportingCatalogName", |
There was a problem hiding this comment.
Nit: spark.conf.set and the catalog lookup run before the try, so the conf would not be unset if the lookup failed. It cannot fail in practice, and the test at L503-L519 uses the same pattern, so this is optional. Could you wrap the body in a single withSQLConf instead, like withRowLevelCatalog (GeneratedColumnWriteSuite.scala:37)? For example, withSQLConf(s"spark.sql.catalog.$customReportingCatalogName" -> classOf[CustomReportingCatalog].getName, SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { ... }. Then the finally only needs to reset the catalog.
There was a problem hiding this comment.
Done, the body is now wrapped in a single withSQLConf like withRowLevelCatalog, and finally only resets the catalog
| // every reference reachable through `arguments()`/`expression()`, while `references()` can | ||
| // return fewer. `ApplyTransform` returns only its top-level arguments, which would miss | ||
| // `f(g(missing))`, and a connector-defined `Transform` can return anything. | ||
| def references(expr: V2Expression): Seq[NamedReference] = expr match { |
There was a problem hiding this comment.
Nit: This local helper has the same name as V2Expression.references(), which the comment above says not to use. So exprs.flatMap(references) at L124 and exprs.flatMap(_.references()) look almost the same but behave differently. Could you rename it, e.g., collectReferences?
There was a problem hiding this comment.
Done, renamed it to collectReferences.
| kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog)) | ||
| .toImmutableArraySeq) | ||
| val keys = kgp.keys() | ||
| val unresolvedColumns = unresolvableColumns(keys.toImmutableArraySeq, relation) |
There was a problem hiding this comment.
Nit: keys is converted with toImmutableArraySeq here and again at L65. Could you convert it once at L53, e.g., val keys = kgp.keys().toImmutableArraySeq? Then this line becomes unresolvableColumns(keys, relation), and the trailing .toImmutableArraySeq at L65 can go.
There was a problem hiding this comment.
Done, keys is converted once.
anuragmantri
left a comment
There was a problem hiding this comment.
Thanks for the review @dongjoon-hyun. Addressed your comments inline.
| // `f(g(missing))`, and a connector-defined `Transform` can return anything. | ||
| def references(expr: V2Expression): Seq[NamedReference] = expr match { | ||
| case ref: NamedReference => Seq(ref) | ||
| case other => other.children().toImmutableArraySeq.flatMap(references) |
There was a problem hiding this comment.
Thanks, good catch. Done. The walk now reads Transform.arguments() and SortOrder.expression(), the same accessors the conversion uses, and falls back to children() only for other expressions such as GeneralScalarExpression. I updated the comment accordingly. I also added connector-defined Transform and SortOrder test classes whose children() disagrees in both directions: a missing reference hidden from children(), which used to fail with Unable to resolve missing, and an extra reference only in children(), which used to drop a valid partitioning.
| case other => other.children().toImmutableArraySeq.flatMap(references) | ||
| } | ||
| exprs.flatMap(references).flatMap { ref => | ||
| val name = ref.fieldNames.toImmutableArraySeq.quoted |
There was a problem hiding this comment.
Done. An empty fieldNames() is now treated as unresolvable before quoted or resolveRefOpt is called, and it is shown as <empty> in the warning. I added an ordering case for it, which used to fail with "Multi-part identifier cannot be empty".
| .toImmutableArraySeq) | ||
| val keys = kgp.keys() | ||
| val unresolvedColumns = unresolvableColumns(keys.toImmutableArraySeq, relation) | ||
| val partitioning = if (unresolvedColumns.nonEmpty) { |
There was a problem hiding this comment.
Done. I added your sentence to the Javadoc of KeyGroupedPartitioning#keys() and SupportsReportOrdering#outputOrdering(), covering only unresolvable references. I also checked the "including metadata columns" part: a reported ordering on a metadata column resolves even when the query does not read it, since LogicalPlan.resolve falls back to the metadata output.
| * is a `KeyedPartitioning` and | ||
| * `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on, each partition | ||
| * contains rows where the key expressions evaluate to a single constant value, so the data | ||
| * is trivially sorted by those expressions within the partition. |
| // every reference reachable through `arguments()`/`expression()`, while `references()` can | ||
| // return fewer. `ApplyTransform` returns only its top-level arguments, which would miss | ||
| // `f(g(missing))`, and a connector-defined `Transform` can return anything. | ||
| def references(expr: V2Expression): Seq[NamedReference] = expr match { |
There was a problem hiding this comment.
Done, renamed it to collectReferences.
| kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog)) | ||
| .toImmutableArraySeq) | ||
| val keys = kgp.keys() | ||
| val unresolvedColumns = unresolvableColumns(keys.toImmutableArraySeq, relation) |
There was a problem hiding this comment.
Done, keys is converted once.
| new GeneralScalarExpression("+", Array[Expression](FieldReference(column), literal(1))) | ||
|
|
||
| private def withCustomReportingTable(f: CustomReportingCatalog => Unit): Unit = { | ||
| spark.conf.set(s"spark.sql.catalog.$customReportingCatalogName", |
There was a problem hiding this comment.
Done, the body is now wrapped in a single withSQLConf like withRowLevelCatalog, and finally only resets the catalog
|
Thanks for re-checking, @peter-toth.
|
|
Thank you for updating, @anuragmantri . |
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing the fifth round, @anuragmantri. I reviewed the latest commit, 7e32de4, and left 8 inline comments. All nine items of the previous round are addressed, and I did not find a regression. The new ChildrenOverridingTransform and ChildrenOverridingSortOrder cases fail with the previous children() walk in both directions, so they pin the fix well. None of the comments is a blocker, and most of them are nits. Here is a summary, ordered by priority.
Docs and comments
KeyGroupedPartitioningL47: the "including metadata columns" part, which I suggested, is imprecise for DELETE, UPDATE, and MERGE, where only the row ID and the required metadata columns can be resolved. The same sentence is inSupportsReportOrdering.
Tests
KeyGroupedPartitioningSuiteL3620: no case has two distinct unresolvable columns, so the column list in the warning is not pinned.KeyGroupedPartitioningSuiteL9541: thereferences()override ofChildrenOverridingTransformis not needed.KeyGroupedPartitioningSuiteL3597: thewarningTextparameter ofcustomReportingJoinis redundant now that the whole message is compared.DataSourceV2SuiteL413:getScanRelationcan be reused. This one is optional.KeyGroupedPartitioningSuiteL3655: the two-argumentsortgives the same sort order. This one is optional.DataSourceV2SuiteL394:spark.sql.sources.v2.bucketing.enabledis already true by default. This one is optional.
Cleanup
V2ScanPartitioningAndOrderingL128:ref.fieldNamesis read twice.
PR description: the second V2ExpressionUtilsSuite bullet still ends at "for an ambiguous", and the testing section does not mention the DataSourceV2Suite test or the mixed cases yet, although your last comment says both were done. Could you check whether the edit was saved? The description becomes the commit message.
Items 1 and 2 are the most relevant ones. The others are minor.
| * Spark resolves the column references in these expressions against the table columns, | ||
| * including metadata columns and columns pruned from the scan output. If any of them cannot | ||
| * be resolved, Spark ignores the whole reported partitioning and logs a warning. |
There was a problem hiding this comment.
Minor: The "including metadata columns" part of the sentence which I suggested is imprecise. For DELETE, UPDATE, and MERGE, RewriteRowLevelCommand.buildRelationWithAttrs replaces the table of the read relation with RowLevelOperationTable, which is not SupportsMetadataColumns, and adds only the row ID and the required metadata columns to its output. So a key or a sort order on another metadata column resolves for a SELECT, like the new index case, but the whole report can be dropped with the warning in these commands. It is not a regression, since such a report failed the command before, and it matters only for a connector which reports keys or sort orders on metadata columns. Could you drop that part here and in SupportsReportOrdering#outputOrdering()? For example,
| * Spark resolves the column references in these expressions against the table columns, | |
| * including metadata columns and columns pruned from the scan output. If any of them cannot | |
| * be resolved, Spark ignores the whole reported partitioning and logs a warning. | |
| * Spark resolves the column references in these expressions against the table columns, | |
| * including columns pruned from the scan output. If any of them cannot be resolved, Spark | |
| * ignores the whole reported partitioning and logs a warning. |
There was a problem hiding this comment.
Thanks, you're right. Done.
| (Seq(bucket(4, "missing")), Some("missing")), | ||
| (Seq(nestedKey("missing")), Some("missing")), | ||
| (Seq(identity("id"), identity("missing")), Some("missing")), | ||
| (Seq(identity("missing"), bucket(4, "missing")), Some("missing")), |
There was a problem hiding this comment.
Minor: Every dropped case here and in the ordering test below expects a single distinct name, so the column list in the warning is not pinned. If unresolvableColumns reported only the first unresolvable column, sorted the names, or used another separator, all the tests would still pass. Could you add a case with two distinct unresolvable columns? For example,
(Seq(identity("missing"), bucket(4, "absent"), identity("missing")),
Some("missing, absent")),It keeps the de-duplication check, and since absent sorts before missing, it pins the first-seen order too.
There was a problem hiding this comment.
Added your case, [identity(missing), bucket(4, absent), identity(missing)], which expects missing, absent. I checked that it fails if unresolvableColumns keeps only the first name.
| override def references(): Array[NamedReference] = | ||
| args.collect { case ref: NamedReference => ref }.toArray |
There was a problem hiding this comment.
Nit: This override is not needed. Transform does not declare references(), so the default Expression.references() applies, and it follows children(). Nothing on the planning path calls it either. Without the override, this class keeps the defaults like ChildrenOverridingSortOrder below, and the case at L3626 (arguments() is [id] and children() is [missing]) also pins that the walk ignores references(). Could you remove it?
There was a problem hiding this comment.
Removed. ChildrenOverridingTransform now keeps the default references(), like ChildrenOverridingSortOrder.
| |JOIN testcat.ns.$table t2 ON t1.id = t2.id | ||
| |""".stripMargin) | ||
|
|
||
| private def customReportingJoin(warningText: String): (SparkPlan, BatchScanExec, Seq[String]) = { |
There was a problem hiding this comment.
Nit: Since ignoredReportWarning pins the whole message, including the kind of the report, this parameter only repeats the kind. A fixed filter works at all three call sites: the partitioning test reports an empty ordering, which is kept without a warning, and the key of the ordering test resolves. Then the warnings.isEmpty checks also cover the other kind. Could you use a fixed filter like _.contains("columns cannot be resolved") and drop the parameter?
There was a problem hiding this comment.
Done. customReportingJoin() now uses the fixed filter _.contains("columns cannot be resolved"), and the parameter is gone.
| val scanRelations = df.queryExecution.optimizedPlan.collect { | ||
| case d: DataSourceV2ScanRelation => d | ||
| } | ||
| assert(scanRelations.length == 1) | ||
| assert(scanRelations.head.ordering.isEmpty) |
There was a problem hiding this comment.
Nit: getScanRelation(df) (L1227) does the same, except for the length == 1 check. In addition, L412 checks only the end of the warning, so the Spark ignores the ordering reported by ... (scan ...) part is not pinned for a relation without a catalog. Since the table name is an anonymous class name here, the whole message would need to be built from the plan, e.g. with scanRelation.name and scanRelation.scan.getClass.getName. This one is optional. Could you use getScanRelation(df) here?
There was a problem hiding this comment.
Done. The test uses getScanRelation(df) and compares the whole warning, built from scanRelation.name and scanRelation.scan.getClass.getName. It also compares the set of warnings now, as KeyGroupedPartitioningSuite does.
| withSQLConf(SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key -> "true") { | ||
| reportingCatalog.reportedKeys = Seq(identity("id")) | ||
| Seq[(SortOrder, String)]( | ||
| (sort(FieldReference("id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST), "id"), |
There was a problem hiding this comment.
Nit: sort(FieldReference("id"), SortDirection.ASCENDING) gives the same sort order, since ASCENDING defaults to NULLS_FIRST, and it would fit each of the five sort calls of this test in one line. This suite uses the three-argument form everywhere else, so this one is optional. Could you use the two-argument form here?
There was a problem hiding this comment.
Done. All the sort calls in this test use the two-argument form.
| } | ||
|
|
||
| test("SPARK-59721: ignore an unresolvable ordering reported with connector-defined classes") { | ||
| withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") { |
There was a problem hiding this comment.
Nit: spark.sql.sources.v2.bucketing.enabled is true by default, and this test checks only the answer, the warning, and the optimized plan's ordering, which do not depend on it. It is fine to keep it to match the neighboring tests, so this one is optional. Could you remove this withSQLConf?
There was a problem hiding this comment.
Removed the withSQLConf.
| case other => other.children().toImmutableArraySeq.flatMap(collectReferences) | ||
| } | ||
| exprs.flatMap(collectReferences).flatMap { ref => | ||
| if (ref.fieldNames.isEmpty) { |
There was a problem hiding this comment.
Nit: ref.fieldNames is read twice now, at L128 and L132, and once more in resolveRefOpt. Like kgp.keys() and scan.outputOrdering(), a connector may build a new array on each call, e.g. FieldReference.fieldNames is parts.toArray. Could you read it once before the check? For example,
val parts = ref.fieldNames.toImmutableArraySeq
if (parts.isEmpty) {
// An empty name can be neither quoted nor resolved: both throw.
Some("<empty>")
} else {
val name = parts.quoted
try {
Option.when(V2ExpressionUtils.resolveRefOpt(ref, relation).isEmpty)(name)
} catch {
// A nested-field extraction error or an ambiguous reference throws instead of returning
// None.
case e: AnalysisException => Some(s"$name (${e.getCondition})")
}
}Option.when is optional.
There was a problem hiding this comment.
Done. fieldNames is read once into parts and reused for the empty check and the quoted name. As in your suggestion, resolveRefOpt still reads it once more inside.
There was a problem hiding this comment.
Thanks @dongjoon-hyun. I addressed all of the round-6 comments in 1ac4916
While re-reviewing the whole PR I also:
- added a case where a two-key ordering has one unresolvable key,
[id, missing], to pin that Spark ignores the whole ordering; - fixed the
PlanMergercomment, which contradicted itself after the earlier edit. - made the
collectReferencescomment precise aboutGeneralScalarExpression.
| * Spark resolves the column references in these expressions against the table columns, | ||
| * including metadata columns and columns pruned from the scan output. If any of them cannot | ||
| * be resolved, Spark ignores the whole reported partitioning and logs a warning. |
There was a problem hiding this comment.
Thanks, you're right. Done.
| (Seq(bucket(4, "missing")), Some("missing")), | ||
| (Seq(nestedKey("missing")), Some("missing")), | ||
| (Seq(identity("id"), identity("missing")), Some("missing")), | ||
| (Seq(identity("missing"), bucket(4, "missing")), Some("missing")), |
There was a problem hiding this comment.
Added your case, [identity(missing), bucket(4, absent), identity(missing)], which expects missing, absent. I checked that it fails if unresolvableColumns keeps only the first name.
| override def references(): Array[NamedReference] = | ||
| args.collect { case ref: NamedReference => ref }.toArray |
There was a problem hiding this comment.
Removed. ChildrenOverridingTransform now keeps the default references(), like ChildrenOverridingSortOrder.
| |JOIN testcat.ns.$table t2 ON t1.id = t2.id | ||
| |""".stripMargin) | ||
|
|
||
| private def customReportingJoin(warningText: String): (SparkPlan, BatchScanExec, Seq[String]) = { |
There was a problem hiding this comment.
Done. customReportingJoin() now uses the fixed filter _.contains("columns cannot be resolved"), and the parameter is gone.
| val scanRelations = df.queryExecution.optimizedPlan.collect { | ||
| case d: DataSourceV2ScanRelation => d | ||
| } | ||
| assert(scanRelations.length == 1) | ||
| assert(scanRelations.head.ordering.isEmpty) |
There was a problem hiding this comment.
Done. The test uses getScanRelation(df) and compares the whole warning, built from scanRelation.name and scanRelation.scan.getClass.getName. It also compares the set of warnings now, as KeyGroupedPartitioningSuite does.
| withSQLConf(SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key -> "true") { | ||
| reportingCatalog.reportedKeys = Seq(identity("id")) | ||
| Seq[(SortOrder, String)]( | ||
| (sort(FieldReference("id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST), "id"), |
There was a problem hiding this comment.
Done. All the sort calls in this test use the two-argument form.
| } | ||
|
|
||
| test("SPARK-59721: ignore an unresolvable ordering reported with connector-defined classes") { | ||
| withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") { |
There was a problem hiding this comment.
Removed the withSQLConf.
| case other => other.children().toImmutableArraySeq.flatMap(collectReferences) | ||
| } | ||
| exprs.flatMap(collectReferences).flatMap { ref => | ||
| if (ref.fieldNames.isEmpty) { |
There was a problem hiding this comment.
Done. fieldNames is read once into parts and reused for the empty check and the quoted name. As in your suggestion, resolveRefOpt still reads it once more inside.
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing the sixth round, @anuragmantri. I reviewed the latest commit, 1ac4916, and left 4 inline comments. All eight items of the previous round are addressed, the PR description is complete now, and I did not find a regression. The new [id, missing] and missing, absent cases look good to me. None of the comments is a blocker, and all of them are nits. Here is a summary, ordered by priority.
Docs and comments
PlanMergerL983: the rewritten comment says again that the partitioning pass keeps the report whenever any key survives, and it no longer says that the ordering pass has no such guard.
Tests
KeyGroupedPartitioningSuiteL3651: the test name reads as if only the unresolvable column is ignored, while the new[id, missing]case pins that the whole ordering is ignored.DataSourceV2SuiteL414: the filter could be"columns cannot be resolved"likecustomReportingJoin(). This one is optional.KeyGroupedPartitioningSuiteL3658: a short comment on whyindexresolves would help.
Item 1 is the most relevant one. The others are minor.
| // GAINS is not a degradation either. For partitioning that is because an input dropped its own | ||
| // only where none of its keys survived in that scan's output (V2ScanPartitioningAndOrdering's | ||
| // partitioning pass keeps the report whenever any key survives), not because the source stopped | ||
| // reporting; the ordering pass has no such guard, so an ordering report is | ||
| // never dropped by pruning and a gained one can only come from the source. Either way, keeping it | ||
| // is exactly the win this merge is after. | ||
| // reporting. Both passes also drop a report that references a column the relation cannot | ||
| // resolve, but that does not depend on pruning, so an ordering report is never dropped by | ||
| // pruning and a gained one can only come from the source. Either way, keeping it is exactly the | ||
| // win this merge is after. |
There was a problem hiding this comment.
Nit: I see why you rewrote this, since "no such guard" right after "unless a reported column cannot be resolved" could be read as covering the unresolvable-column drop too. However, the new text has two gaps. First, "only where none of its keys survived" and "keeps the report whenever any key survives" are literally false again, e.g. [identity(id), identity(missing)] drops the whole report although id survives, and the "also" in the next sentence contradicts them. Second, the removed "the ordering pass has no such guard" was the reason why an ordering report is never dropped by pruning. The new premise, that the unresolvable-column drop does not depend on pruning, holds for the partitioning pass too. Could you state the unresolvable-column drop first and keep the original argument for pruning? For example,
| // GAINS is not a degradation either. For partitioning that is because an input dropped its own | |
| // only where none of its keys survived in that scan's output (V2ScanPartitioningAndOrdering's | |
| // partitioning pass keeps the report whenever any key survives), not because the source stopped | |
| // reporting; the ordering pass has no such guard, so an ordering report is | |
| // never dropped by pruning and a gained one can only come from the source. Either way, keeping it | |
| // is exactly the win this merge is after. | |
| // reporting. Both passes also drop a report that references a column the relation cannot | |
| // resolve, but that does not depend on pruning, so an ordering report is never dropped by | |
| // pruning and a gained one can only come from the source. Either way, keeping it is exactly the | |
| // win this merge is after. | |
| // GAINS is not a degradation either. Both V2ScanPartitioningAndOrdering passes drop a report | |
| // that references a column the relation cannot resolve, but that does not depend on pruning. | |
| // Beyond that, an input dropped its partitioning only where none of its keys survived in that | |
| // scan's output (the partitioning pass keeps the report whenever any key survives), not because | |
| // the source stopped reporting; the ordering pass has no such guard, so an ordering report is | |
| // never dropped by pruning and a gained one can only come from the source. Either way, keeping | |
| // it is exactly the win this merge is after. |
There was a problem hiding this comment.
Thanks, you're right on both points. I took your wording: it states the unresolvable-column drop first and brings back the original pruning argument, including "the ordering pass has no such guard".
| } | ||
| } | ||
|
|
||
| test("SPARK-59721: an unresolvable reported ordering column is ignored") { |
There was a problem hiding this comment.
Nit: The [id, missing] case which you added in this round pins that the whole reported ordering is ignored, including the resolvable id, as the Javadoc of SupportsReportOrdering#outputOrdering() says. The test name still reads as if only the unresolvable column is ignored. Could you rename it? For example,
| test("SPARK-59721: an unresolvable reported ordering column is ignored") { | |
| test("SPARK-59721: a reported ordering with an unresolvable column is ignored") { |
There was a problem hiding this comment.
Renamed it to "a reported ordering with an unresolvable column is ignored". I renamed the partitioning test to match, "a reported partitioning with an unresolvable key is ignored", since its [id, missing] case also drops the whole report.
| s"(scan ${scanRelation.scan.getClass.getName}) because the ordering columns " + | ||
| "cannot be resolved: missing." | ||
| val warnings = logAppender.loggingEvents.map(_.getMessage.getFormattedMessage) | ||
| .filter(_.contains("ordering columns cannot be resolved")) |
There was a problem hiding this comment.
Nit: Since expectedWarning pins the whole message including the kind of the report, the filter could be the same as customReportingJoin() in KeyGroupedPartitioningSuite. Then the set comparison also checks that the resolvable partitionKeys=i produces no warning. This one is optional. Could you widen it?
| .filter(_.contains("ordering columns cannot be resolved")) | |
| .filter(_.contains("columns cannot be resolved")) |
There was a problem hiding this comment.
Done. The filter is now _.contains("columns cannot be resolved").
| Seq[(SortOrder, String)]( | ||
| (sort(FieldReference("id"), SortDirection.ASCENDING), "id"), | ||
| (sort(FieldReference("ID"), SortDirection.ASCENDING), "ID"), | ||
| (sort(FieldReference("index"), SortDirection.ASCENDING), "index"), |
There was a problem hiding this comment.
Nit: The reporting table has only id, data and s, so it is not obvious why index resolves here. Since the Javadoc no longer mentions metadata columns, could you add a short comment? For example,
| (sort(FieldReference("index"), SortDirection.ASCENDING), "index"), | |
| // `index` is a metadata column of InMemoryBaseTable. | |
| (sort(FieldReference("index"), SortDirection.ASCENDING), "index"), |
There was a problem hiding this comment.
Added the comment.
anuragmantri
left a comment
There was a problem hiding this comment.
Thanks @dongjoon-hyun. I addressed the four round-7 nits.
| // GAINS is not a degradation either. For partitioning that is because an input dropped its own | ||
| // only where none of its keys survived in that scan's output (V2ScanPartitioningAndOrdering's | ||
| // partitioning pass keeps the report whenever any key survives), not because the source stopped | ||
| // reporting; the ordering pass has no such guard, so an ordering report is | ||
| // never dropped by pruning and a gained one can only come from the source. Either way, keeping it | ||
| // is exactly the win this merge is after. | ||
| // reporting. Both passes also drop a report that references a column the relation cannot | ||
| // resolve, but that does not depend on pruning, so an ordering report is never dropped by | ||
| // pruning and a gained one can only come from the source. Either way, keeping it is exactly the | ||
| // win this merge is after. |
There was a problem hiding this comment.
Thanks, you're right on both points. I took your wording: it states the unresolvable-column drop first and brings back the original pruning argument, including "the ordering pass has no such guard".
| } | ||
| } | ||
|
|
||
| test("SPARK-59721: an unresolvable reported ordering column is ignored") { |
There was a problem hiding this comment.
Renamed it to "a reported ordering with an unresolvable column is ignored". I renamed the partitioning test to match, "a reported partitioning with an unresolvable key is ignored", since its [id, missing] case also drops the whole report.
| s"(scan ${scanRelation.scan.getClass.getName}) because the ordering columns " + | ||
| "cannot be resolved: missing." | ||
| val warnings = logAppender.loggingEvents.map(_.getMessage.getFormattedMessage) | ||
| .filter(_.contains("ordering columns cannot be resolved")) |
There was a problem hiding this comment.
Done. The filter is now _.contains("columns cannot be resolved").
| Seq[(SortOrder, String)]( | ||
| (sort(FieldReference("id"), SortDirection.ASCENDING), "id"), | ||
| (sort(FieldReference("ID"), SortDirection.ASCENDING), "ID"), | ||
| (sort(FieldReference("index"), SortDirection.ASCENDING), "index"), |
There was a problem hiding this comment.
Added the comment.
…erence unresolvable columns ### What changes were proposed in this pull request? When a V2 scan reports a `KeyGroupedPartitioning` or an `ordering` that references a column the relation can't resolve, the query used to fail with `Unable to resolve X given [...]`. This PR makes `V2ScanPartitioningAndOrdering` check the columns referenced by the reported partition keys and ordering (recursively, including nested transform arguments) against the relation before converting them. If any can't be resolved, including a missing nested field, an ambiguous name, or an empty name, the rule logs a warning naming the relation, the scan class, and those columns, and ignores the report. The partitioning falls back to `UnknownPartitioning`, and the ordering is dropped. This PR adds a `resolveRefOpt`, and makes `resolveRef` a thin wrapper around it. `resolveRefOpt` still throws for a missing nested field or an ambiguous name, so the rule treats those errors as unresolvable and includes the error condition in the warning. The `toCatalyst*` methods are unchanged. Keys whose columns all resolve take the existing conversion path unchanged, so an unsupported expression type (e.g. `id + 1`) or a nested transform whose function can't be loaded still fails the query as before. Because the whole report is checked before any key is converted, a report that also references an unresolvable column falls back with a warning instead, e.g. `[id + 1, identity(missing)]`. The Javadoc of `KeyGroupedPartitioning#keys()` and `SupportsReportOrdering#outputOrdering()` now says that Spark resolves the column references against the table columns, including columns pruned from the scan output, and ignores the whole report if any of them can't be resolved. The docs of `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` now mention that an ignored ordering can be replaced by one derived from the partition keys. ### Why are the changes needed? Found while reviewing SPARK-58111 ([PR #55518](#55518)). This is a general, pre-existing bug in how `V2ScanPartitioningAndOrdering` handles reported partitioning and ordering, not something introduced by #55518. ### Does this PR introduce _any_ user-facing change? Yes. A query against a data source whose scan reports a `KeyGroupedPartitioning` with a key that can't be resolved previously failed outright. It now succeeds, with partitioning degraded to `UnknownPartitioning` instead, and Spark logs a warning naming the relation, the scan class, and the columns that can't be resolved. The same applies to an ambiguous or empty column name. A report that combines an unsupported expression with an unresolvable column, e.g. `missing + 1` or `[id + 1, identity(missing)]`, used to fail with `_LEGACY_ERROR_TEMP_3054` and now falls back with a warning. An unsupported expression over resolvable columns still fails as before. If the scan keeps its partitioning but its reported ordering is ignored, Spark derives the ordering from the partition keys when `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on (the default). ### How was this patch tested? New tests in `V2ExpressionUtilsSuite`: - `resolveRefOpt` returns `None` for an unresolvable reference. - `resolveRefOpt` throws `FIELD_NOT_FOUND` for a missing nested field and `AMBIGUOUS_REFERENCE` for an ambiguous reference. New tests in `KeyGroupedPartitioningSuite`: - A join against a table whose scan reports unresolvable partition keys succeeds, plans a shuffle instead of a storage-partitioned join, and logs a warning naming the relation, the scan class, and only the unresolvable columns. The cases are: - a `bucket` key, a nested transform key, and a multi-key partitioning with one bad key; - duplicate references, and two distinct unresolvable columns; - a missing nested field; - an unsupported expression next to or over an unresolvable column (`[id + 1, identity(missing)]`, `missing + 1`); - a connector-defined transform that hides the reference from `children()`. Resolvable keys (`id`, case-insensitive `ID`, and a connector-defined transform with an extra reference only in `children()`) keep the partitioning and plan a storage-partitioned join without a shuffle. - A reported ordering that can't be resolved is ignored as a whole with a warning, and the scan still derives `id ASC` from its kept partitioning. The cases are a column that doesn't exist, one hidden from a connector-defined sort order's `children()`, an empty name, and a two-key ordering with one unresolvable key (`[id, missing]`). Resolvable orderings (`id`, case-insensitive `ID`, the metadata column `index`, and a connector-defined sort order with an extra reference only in `children()`) are kept without a warning. - A reported partition key of an unsupported expression type (`id + 1`), or a nested transform whose function can't be loaded (`f(g(id))`), still fails the query with `_LEGACY_ERROR_TEMP_3054`. New test in `DataSourceV2Suite`: - An unresolvable ordering reported through connector-defined `SortOrder`, `Transform`, and `NamedReference` classes (`OrderAndPartitionAwareDataSource` and its Java counterpart) is ignored with a warning, and the scan relation keeps no ordering. Also ran `V2ExpressionUtilsSuite`, `KeyGroupedPartitioningSuite`, `WriteDistributionAndOrderingSuite`, `DataSourceV2Suite`, `MergeSubplansSuite` (comment update only), `SQLConfSuite`, and `ProjectedOrderingAndPartitioningSuite` to confirm no regression in the partitioning/SPJ, ordering, and write paths this change touches. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Sonnet 5, Opus 5.5) Verified manually by me. Closes #58979 from anuragmantri/v2expr-opt-resolve-degrade. Authored-by: Anurag Mantripragada <amantripragada@apple.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit 9de605f) Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
|
Thank you, @anuragmantri and @peter-toth ! |
What changes were proposed in this pull request?
When a V2 scan reports a
KeyGroupedPartitioningor anorderingthat references a column the relation can't resolve, the query used to fail withUnable to resolve X given [...]. This PR makesV2ScanPartitioningAndOrderingcheck the columns referenced by the reported partition keys and ordering (recursively, including nested transform arguments) against the relation before converting them. If any can't be resolved, including a missing nested field, an ambiguous name, or an empty name, the rule logs a warning naming the relation, the scan class, and those columns, and ignores the report. The partitioning falls back toUnknownPartitioning, and the ordering is dropped.This PR adds a
resolveRefOpt, and makesresolveRefa thin wrapper around it.resolveRefOptstill throws for a missing nested field or an ambiguous name, so the rule treats those errors as unresolvable and includes the error condition in the warning. ThetoCatalyst*methods are unchanged.Keys whose columns all resolve take the existing conversion path unchanged, so an unsupported expression type (e.g.
id + 1) or a nested transform whose function can't be loaded still fails the query as before. Because the whole report is checked before any key is converted, a report that also references an unresolvable column falls back with a warning instead, e.g.[id + 1, identity(missing)].The Javadoc of
KeyGroupedPartitioning#keys()andSupportsReportOrdering#outputOrdering()now says that Spark resolves the column references against the table columns, including columns pruned from the scan output, and ignores the whole report if any of them can't be resolved. The docs ofspark.sql.sources.v2.bucketing.partitionKeyOrdering.enablednow mention that an ignored ordering can be replaced by one derived from the partition keys.Why are the changes needed?
Found while reviewing SPARK-58111 (PR #55518). This is a general, pre-existing bug in how
V2ScanPartitioningAndOrderinghandles reported partitioning and ordering, not something introduced by #55518.Does this PR introduce any user-facing change?
Yes. A query against a data source whose scan reports a
KeyGroupedPartitioningwith a key that can't be resolved previously failed outright. It now succeeds, with partitioning degraded toUnknownPartitioninginstead, and Spark logs a warning naming the relation, the scan class, and the columns that can't be resolved.The same applies to an ambiguous or empty column name. A report that combines an unsupported expression with an unresolvable column, e.g.
missing + 1or[id + 1, identity(missing)], used to fail with_LEGACY_ERROR_TEMP_3054and now falls back with a warning. An unsupported expression over resolvable columns still fails as before.If the scan keeps its partitioning but its reported ordering is ignored, Spark derives the ordering from the partition keys when
spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabledis on (the default).How was this patch tested?
New tests in
V2ExpressionUtilsSuite:resolveRefOptreturnsNonefor an unresolvable reference.resolveRefOptthrowsFIELD_NOT_FOUNDfor a missing nested field andAMBIGUOUS_REFERENCEfor an ambiguous reference.New tests in
KeyGroupedPartitioningSuite:A join against a table whose scan reports unresolvable partition keys succeeds, plans a shuffle instead of a storage-partitioned join, and logs a warning naming the relation, the scan class, and only the unresolvable columns. The cases are:
bucketkey, a nested transform key, and a multi-key partitioning with one bad key;[id + 1, identity(missing)],missing + 1);children().Resolvable keys (
id, case-insensitiveID, and a connector-defined transform with an extra reference only inchildren()) keep the partitioning and plan a storage-partitioned join without a shuffle.A reported ordering that can't be resolved is ignored as a whole with a warning, and the scan still derives
id ASCfrom its kept partitioning. The cases are a column that doesn't exist, one hidden from a connector-defined sort order'schildren(), an empty name, and a two-key ordering with one unresolvable key ([id, missing]). Resolvable orderings (id, case-insensitiveID, the metadata columnindex, and a connector-defined sort order with an extra reference only inchildren()) are kept without a warning.A reported partition key of an unsupported expression type (
id + 1), or a nested transform whose function can't be loaded (f(g(id))), still fails the query with_LEGACY_ERROR_TEMP_3054.New test in
DataSourceV2Suite:SortOrder,Transform, andNamedReferenceclasses (OrderAndPartitionAwareDataSourceand its Java counterpart) is ignored with a warning, and the scan relation keeps no ordering.Also ran
V2ExpressionUtilsSuite,KeyGroupedPartitioningSuite,WriteDistributionAndOrderingSuite,DataSourceV2Suite,MergeSubplansSuite(comment update only),SQLConfSuite, andProjectedOrderingAndPartitioningSuiteto confirm no regression in the partitioning/SPJ, ordering, and write paths this change touches.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Sonnet 5, Opus 5.5)
Verified manually by me.