dongjoon-hyun commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4158298381
##########
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:
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`?
##########
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:
**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()`:
```java
* <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.
##########
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:
**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`,
```scala
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.
##########
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:
Nit: The reflow stops here and leaves a short line. Could you reflow the
rest of the paragraph?
```suggestion
* 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.
```
##########
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",
+ classOf[CustomReportingCatalog].getName)
+ val reportingCatalog =
spark.sessionState.catalogManager.catalog(customReportingCatalogName)
+ .asInstanceOf[CustomReportingCatalog]
+ try {
+ val reportingColumns = Array(
+ Column.create("id", IntegerType),
+ Column.create("data", StringType),
+ Column.create("s", new StructType().add("x", IntegerType)))
+ createTable(table, reportingColumns, Array(identity("id")), catalog =
reportingCatalog)
+ sql(s"INSERT INTO $customReportingTableName VALUES " +
+ "(1, 'aa', named_struct('x', 10)), (2, 'bb', named_struct('x', 20))")
+ createTable(table, columns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$table VALUES " +
+ "(1, 'cc', cast('2020-01-03' as timestamp)), " +
+ "(2, 'dd', cast('2020-01-04' as timestamp))")
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ f(reportingCatalog)
+ }
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ spark.conf.unset(s"spark.sql.catalog.$customReportingCatalogName")
+ }
+ }
+
+ private def customReportingJoinDf(): DataFrame = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingTableName t1
+ |JOIN testcat.ns.$table t2 ON t1.id = t2.id
+ |""".stripMargin)
+
+ private def customReportingJoin(warningText: String): (SparkPlan,
BatchScanExec, Seq[String]) = {
+ val df = customReportingJoinDf()
+ val logAppender = new LogAppender("custom reported partitioning or
ordering")
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ checkAnswer(df, Seq(Row("aa", "cc"), Row("bb", "dd")))
+ }
+ val warnings =
logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ .filter(_.contains(warningText)).toSeq
+ val plan = df.queryExecution.executedPlan
+ val reportingScans =
collectScans(plan).filter(_.scan.isInstanceOf[CustomReportingScan])
+ assert(reportingScans.length == 1)
+ (plan, reportingScans.head, warnings)
+ }
+
+ test("SPARK-59721: an unresolvable reported partition key falls back to
unknown partitioning") {
+ val nestedTransform =
+ ApplyTransform("f", Seq(ApplyTransform("g",
Seq(FieldReference("missing")))))
Review Comment:
Nit: The same nested key is built again at L3676 with `id`. Could you add a
helper next to `plusOneKey`? For example,
```scala
private def nestedKey(column: String): Expression =
ApplyTransform("f", Seq(ApplyTransform("g",
Seq(FieldReference(column)))))
```
Then both become one-line cases like the `plusOneKey` ones.
##########
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:
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.
##########
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:
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`?
##########
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",
+ classOf[CustomReportingCatalog].getName)
+ val reportingCatalog =
spark.sessionState.catalogManager.catalog(customReportingCatalogName)
+ .asInstanceOf[CustomReportingCatalog]
+ try {
+ val reportingColumns = Array(
+ Column.create("id", IntegerType),
+ Column.create("data", StringType),
+ Column.create("s", new StructType().add("x", IntegerType)))
+ createTable(table, reportingColumns, Array(identity("id")), catalog =
reportingCatalog)
+ sql(s"INSERT INTO $customReportingTableName VALUES " +
+ "(1, 'aa', named_struct('x', 10)), (2, 'bb', named_struct('x', 20))")
+ createTable(table, columns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$table VALUES " +
+ "(1, 'cc', cast('2020-01-03' as timestamp)), " +
+ "(2, 'dd', cast('2020-01-04' as timestamp))")
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ f(reportingCatalog)
+ }
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ spark.conf.unset(s"spark.sql.catalog.$customReportingCatalogName")
+ }
+ }
+
+ private def customReportingJoinDf(): DataFrame = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingTableName t1
+ |JOIN testcat.ns.$table t2 ON t1.id = t2.id
+ |""".stripMargin)
+
+ private def customReportingJoin(warningText: String): (SparkPlan,
BatchScanExec, Seq[String]) = {
+ val df = customReportingJoinDf()
+ val logAppender = new LogAppender("custom reported partitioning or
ordering")
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ checkAnswer(df, Seq(Row("aa", "cc"), Row("bb", "dd")))
+ }
+ val warnings =
logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ .filter(_.contains(warningText)).toSeq
+ val plan = df.queryExecution.executedPlan
+ val reportingScans =
collectScans(plan).filter(_.scan.isInstanceOf[CustomReportingScan])
+ assert(reportingScans.length == 1)
+ (plan, reportingScans.head, warnings)
+ }
+
+ test("SPARK-59721: an unresolvable reported partition key falls back to
unknown partitioning") {
+ val nestedTransform =
+ 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])](
+ (Seq(identity("id")), None),
+ (Seq(bucket(4, "missing")), Some("missing")),
+ (Seq(nestedTransform), Some("missing")),
+ (Seq(identity("id"), identity("missing")), Some("missing")),
+ (Seq(identity("missing"), bucket(4, "missing")), Some("missing")),
+ (Seq(plusOneKey("id"), identity("missing")), Some("missing")),
+ (Seq(plusOneKey("missing")), Some("missing")),
+ (Seq(identity("s.missing")), Some("s.missing (FIELD_NOT_FOUND)")))
+ withCustomReportingTable { reportingCatalog =>
+ cases.foreach { case (keys, unresolvedColumns) =>
+ reportingCatalog.reportedKeys = keys
+ val (plan, reportingScan, warnings) =
+ customReportingJoin("partition key columns cannot be resolved")
+ val keysString = keys.map(_.describe()).mkString(", ")
+ unresolvedColumns match {
+ case Some(columnsString) =>
+ assert(warnings.nonEmpty, s"no warning for keys [$keysString]")
+ assert(warnings.forall(_.contains(s"reported by
$customReportingTableName ")), warnings)
Review 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,
```scala
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
```scala
assert(warnings.toSet ==
Set(ignoredReportWarning("KeyGroupedPartitioning", "partition
key", columnsString)),
keysString)
```
`toSet` also absorbs the repeated warning from the idempotence check of the
`Once` batch.
##########
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:
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.
--
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]
