dongjoon-hyun commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4164038367
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/partitioning/KeyGroupedPartitioning.java:
##########
@@ -43,6 +43,10 @@ public KeyGroupedPartitioning(Expression[] keys, int
numPartitions) {
/**
* Returns the partition transform expressions for this partitioning.
+ * <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.
Review 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,
```suggestion
* 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.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3545,6 +3551,161 @@ 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 nestedKey(column: String): Expression =
+ ApplyTransform("f", Seq(ApplyTransform("g", Seq(FieldReference(column)))))
+ 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."
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ withSQLConf(
+ s"spark.sql.catalog.$customReportingCatalogName" ->
classOf[CustomReportingCatalog].getName,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ 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))")
+ f(reportingCatalog)
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ }
+ }
+ }
+
+ 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]) = {
Review 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?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9318,3 +9479,77 @@ class KeyGroupedPartitioningCatalystRuntimeFilterSuite
catalog.clearTables()
}
}
+
+/**
+ * An in-memory catalog whose scans report the partitioning and ordering set
on the catalog instead
+ * of the table's own, so a test can report keys on columns the table does not
have.
+ */
+class CustomReportingCatalog extends InMemoryTableCatalog {
+ var reportedKeys: Seq[Expression] = Seq.empty
+ var reportedOrdering: Seq[SortOrder] = Seq.empty
+
+ // scalastyle:off argcount
+ override protected def newInMemoryTable(
+ name: String,
+ columns: Array[Column],
+ partitioning: Array[Transform],
+ properties: java.util.Map[String, String],
+ constraints: Array[Constraint],
+ distribution: Distribution,
+ ordering: Array[SortOrder],
+ requiredNumPartitions: Option[Int],
+ advisoryPartitionSize: Option[Long],
+ distributionStrictlyRequired: Boolean,
+ numRowsPerSplit: Int,
+ id: String): InMemoryBaseTable = {
+ // scalastyle:on argcount
+ new InMemoryTable(name, columns, partitioning, properties, constraints,
distribution,
+ ordering, requiredNumPartitions, advisoryPartitionSize,
distributionStrictlyRequired,
+ numRowsPerSplit, id) {
+ override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder =
+ new InMemoryScanBuilder(schema(), options) {
+ override def build(): Scan =
+ CustomReportingScan(super.build(), reportedKeys, reportedOrdering)
+ }
+ }
+ }
+}
+
+case class CustomReportingScan(
+ inner: Scan,
+ keys: Seq[Expression],
+ ordering: Seq[SortOrder]) extends SupportsReportPartitioning with
SupportsReportOrdering {
+ override def readSchema(): StructType = inner.readSchema()
+ override def toBatch: Batch = inner.toBatch
+ override def description(): String = inner.description()
+ override def outputPartitioning(): Partitioning = new KeyGroupedPartitioning(
+ keys.toArray,
+
inner.asInstanceOf[SupportsReportPartitioning].outputPartitioning().numPartitions())
+ override def outputOrdering(): Array[SortOrder] = ordering.toArray
+}
+
+/**
+ * A connector-defined transform whose `children()` differs from its
`arguments()`, which the
+ * default `Transform.children()` returns.
+ */
+case class ChildrenOverridingTransform(
+ name: String,
+ args: Seq[Expression],
+ childExprs: Seq[Expression]) extends Transform {
+ override def arguments(): Array[Expression] = args.toArray
+ override def children(): Array[Expression] = childExprs.toArray
+ override def references(): Array[NamedReference] =
+ args.collect { case ref: NamedReference => ref }.toArray
Review 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?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3545,6 +3551,161 @@ 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 nestedKey(column: String): Expression =
+ ApplyTransform("f", Seq(ApplyTransform("g", Seq(FieldReference(column)))))
+ 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."
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ withSQLConf(
+ s"spark.sql.catalog.$customReportingCatalogName" ->
classOf[CustomReportingCatalog].getName,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ 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))")
+ f(reportingCatalog)
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ }
+ }
+ }
+
+ 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") {
+ // 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(identity("ID")), None),
+ (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")),
Review 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,
```scala
(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.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala:
##########
@@ -389,6 +390,36 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-59721: ignore an unresolvable ordering reported with
connector-defined classes") {
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") {
+ Seq(
+ classOf[OrderAndPartitionAwareDataSource],
+ classOf[JavaOrderAndPartitionAwareDataSource]
+ ).foreach { cls =>
+ withClue(cls.getName) {
+ val df = spark.read
+ .option("partitionKeys", "i")
+ .option("orderKeys", "missing")
+ .format(cls.getName)
+ .load()
+ val logAppender = new LogAppender("unresolvable reported ordering")
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ checkAnswer(df, Seq(Row(1, 4), Row(1, 5), Row(3, 5), Row(2, 6),
Row(4, 1), Row(4, 2)))
+ }
+ val warnings =
logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ .filter(_.contains("ordering columns cannot be resolved"))
+ assert(warnings.nonEmpty)
+ assert(warnings.forall(_.contains("cannot be resolved: missing.")),
warnings)
+ val scanRelations = df.queryExecution.optimizedPlan.collect {
+ case d: DataSourceV2ScanRelation => d
+ }
+ assert(scanRelations.length == 1)
+ assert(scanRelations.head.ordering.isEmpty)
Review 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?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala:
##########
@@ -389,6 +390,36 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-59721: ignore an unresolvable ordering reported with
connector-defined classes") {
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") {
Review 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`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,12 +88,60 @@ 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(", "))}.")
+ // Use None, not Some(Nil): only None lets
DataSourceV2ScanExecBase.outputOrdering derive an
+ // ordering 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 the accessors the conversion reads, `Transform.arguments()` and
+ // `SortOrder.expression()`, instead of `V2Expression.references()` or
`children()`: a connector
+ // can override those to disagree with the conversion, and
`ApplyTransform.references()` returns
+ // only its top-level arguments, which would miss `f(g(missing))`. Other
expressions, such as
+ // `GeneralScalarExpression`, go through `children()`.
+ def collectReferences(expr: V2Expression): Seq[NamedReference] = expr
match {
+ case ref: NamedReference => Seq(ref)
+ case t: Transform =>
t.arguments().toImmutableArraySeq.flatMap(collectReferences)
+ case s: V2SortOrder => collectReferences(s.expression())
+ case other =>
other.children().toImmutableArraySeq.flatMap(collectReferences)
+ }
+ exprs.flatMap(collectReferences).flatMap { ref =>
+ if (ref.fieldNames.isEmpty) {
Review 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,
```scala
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3545,6 +3551,161 @@ 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 nestedKey(column: String): Expression =
+ ApplyTransform("f", Seq(ApplyTransform("g", Seq(FieldReference(column)))))
+ 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."
+
+ private def withCustomReportingTable(f: CustomReportingCatalog => Unit):
Unit = {
+ withSQLConf(
+ s"spark.sql.catalog.$customReportingCatalogName" ->
classOf[CustomReportingCatalog].getName,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ 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))")
+ f(reportingCatalog)
+ } finally {
+ reportingCatalog.reportedKeys = Seq.empty
+ reportingCatalog.reportedOrdering = Seq.empty
+ reportingCatalog.clearTables()
+ }
+ }
+ }
+
+ 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") {
+ // 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(identity("ID")), None),
+ (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")),
+ (Seq(plusOneKey("id"), identity("missing")), Some("missing")),
+ (Seq(plusOneKey("missing")), Some("missing")),
+ (Seq(identity("s.missing")), Some("s.missing (FIELD_NOT_FOUND)")),
+ (Seq(ChildrenOverridingTransform("identity",
Seq(FieldReference("missing")), Seq.empty)),
+ Some("missing")),
+ (Seq(ChildrenOverridingTransform("identity", Seq(FieldReference("id")),
+ Seq(FieldReference("missing")))), None))
+ 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.toSet ==
+ Set(ignoredReportWarning("KeyGroupedPartitioning", "partition
key", columnsString)),
+ keysString)
+ assert(reportingScan.keyGroupedPartitioning.isEmpty, keysString)
+ assert(collectShuffles(plan).nonEmpty, keysString)
+ case None =>
+ assert(warnings.isEmpty, warnings)
+ assert(reportingScan.keyGroupedPartitioning.isDefined, keysString)
+ assert(collectShuffles(plan).isEmpty, keysString)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59721: an unresolvable reported ordering column is ignored") {
+ withCustomReportingTable { reportingCatalog =>
+ 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"),
Review 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?
--
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]
