anuragmantri commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4161658922
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -47,9 +50,20 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
if d.keyGroupedPartitioning.isEmpty =>
val catalystPartitioning = scan.outputPartitioning() match {
case kgp: KeyGroupedPartitioning =>
- val partitioning = sequenceToOption(
- kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
- .toImmutableArraySeq)
+ val keys = kgp.keys()
+ val unresolvedColumns =
unresolvableColumns(keys.toImmutableArraySeq, relation)
+ val partitioning = if (unresolvedColumns.nonEmpty) {
Review 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.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,12 +89,51 @@ 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, _) =>
- // The ordering is kept as reported, even where it references columns
pruned out of the scan
- // output: truncating it here would also drop the sort orders on a
partition key past a
- // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
- // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val reportedOrdering = scan.outputOrdering()
+ val unresolvedColumns =
unresolvableColumns(reportedOrdering.toImmutableArraySeq, relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the ordering reported by ${MDC(RELATION_NAME,
relation.name)} " +
+ log"(scan ${MDC(CLASS_NAME, scan.getClass.getName)}) because the
ordering columns " +
+ log"cannot be resolved: ${MDC(COLUMN_NAMES,
unresolvedColumns.mkString(", "))}.")
+ // With no reported ordering, DataSourceV2ScanExecBase.outputOrdering
may still derive one
+ // from a kept key-grouped partitioning, which does not depend on the
dropped report.
+ d.copy(ordering = None)
+ } else {
+ // The ordering is kept as reported, even where it references columns
pruned out of the
+ // scan output: truncating it here would also drop the sort orders on
a partition key past a
+ // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
+ // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // Walk `children()` instead of calling `V2Expression.references()`: the
conversion resolves
+ // 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 {
+ case ref: NamedReference => Seq(ref)
+ case other => other.children().toImmutableArraySeq.flatMap(references)
Review 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.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3545,6 +3551,142 @@ class KeyGroupedPartitioningSuite
Row(1, "bb", "bb"), Row(2, "cc", "cc")))
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+ private def plusOneKey(column: String): Expression =
+ new GeneralScalarExpression("+", Array[Expression](FieldReference(column),
literal(1)))
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ spark.conf.set(s"spark.sql.catalog.$customReportingCatalogName",
Review Comment:
Done, the body is now wrapped in a single `withSQLConf` like
`withRowLevelCatalog`, and `finally` only resets the catalog
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,12 +89,51 @@ 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, _) =>
- // The ordering is kept as reported, even where it references columns
pruned out of the scan
- // output: truncating it here would also drop the sort orders on a
partition key past a
- // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
- // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val reportedOrdering = scan.outputOrdering()
+ val unresolvedColumns =
unresolvableColumns(reportedOrdering.toImmutableArraySeq, relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the ordering reported by ${MDC(RELATION_NAME,
relation.name)} " +
+ log"(scan ${MDC(CLASS_NAME, scan.getClass.getName)}) because the
ordering columns " +
+ log"cannot be resolved: ${MDC(COLUMN_NAMES,
unresolvedColumns.mkString(", "))}.")
+ // With no reported ordering, DataSourceV2ScanExecBase.outputOrdering
may still derive one
+ // from a kept key-grouped partitioning, which does not depend on the
dropped report.
+ d.copy(ordering = None)
+ } else {
+ // The ordering is kept as reported, even where it references columns
pruned out of the
+ // scan output: truncating it here would also drop the sort orders on
a partition key past a
+ // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
+ // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // Walk `children()` instead of calling `V2Expression.references()`: the
conversion resolves
+ // 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 {
+ case ref: NamedReference => Seq(ref)
+ case other => other.children().toImmutableArraySeq.flatMap(references)
+ }
+ exprs.flatMap(references).flatMap { ref =>
+ val name = ref.fieldNames.toImmutableArraySeq.quoted
Review 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".
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -47,9 +50,20 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
if d.keyGroupedPartitioning.isEmpty =>
val catalystPartitioning = scan.outputPartitioning() match {
case kgp: KeyGroupedPartitioning =>
- val partitioning = sequenceToOption(
- kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
- .toImmutableArraySeq)
+ val keys = kgp.keys()
+ val unresolvedColumns =
unresolvableColumns(keys.toImmutableArraySeq, relation)
Review Comment:
Done, `keys` is converted once.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,12 +89,51 @@ 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, _) =>
- // The ordering is kept as reported, even where it references columns
pruned out of the scan
- // output: truncating it here would also drop the sort orders on a
partition key past a
- // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
- // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val reportedOrdering = scan.outputOrdering()
+ val unresolvedColumns =
unresolvableColumns(reportedOrdering.toImmutableArraySeq, relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the ordering reported by ${MDC(RELATION_NAME,
relation.name)} " +
+ log"(scan ${MDC(CLASS_NAME, scan.getClass.getName)}) because the
ordering columns " +
+ log"cannot be resolved: ${MDC(COLUMN_NAMES,
unresolvedColumns.mkString(", "))}.")
+ // With no reported ordering, DataSourceV2ScanExecBase.outputOrdering
may still derive one
+ // from a kept key-grouped partitioning, which does not depend on the
dropped report.
+ d.copy(ordering = None)
+ } else {
+ // The ordering is kept as reported, even where it references columns
pruned out of the
+ // scan output: truncating it here would also drop the sort orders on
a partition key past a
+ // pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
+ // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // Walk `children()` instead of calling `V2Expression.references()`: the
conversion resolves
+ // 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 {
Review Comment:
Done, renamed it to `collectReferences`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -135,12 +135,13 @@ trait DataSourceV2ScanExecBase
/**
* Returns the output ordering for this scan. When the source reports
ordering via
- * `SupportsReportOrdering`, that ordering may reference columns pruned out
of the scan output
- * (see V2ScanPartitioningAndOrdering), while a consumer binds it against
the output, e.g. the
- * k-way merge of `GroupPartitionsExec`. Ordering is prefix-based, so the
leading run of sort
- * orders over the output is kept and the rest is dropped, except the sort
orders on a partition
- * key when the output partitioning is a `KeyedPartitioning`: each partition
holds a single key,
- * so those still hold. Otherwise, when the output partitioning is a
`KeyedPartitioning` and
+ * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it,
that ordering may
+ * reference columns pruned out of the scan output, while a consumer binds
it against the
+ * output, e.g. the k-way merge of `GroupPartitionsExec`. Ordering is
prefix-based, so the
+ * leading run of sort orders over the output is kept and the rest is
dropped, except the sort
+ * orders on a partition key when the output partitioning is a
`KeyedPartitioning`: each
+ * partition holds a single key, so those still hold. Otherwise, when the
output partitioning
+ * 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.
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
