dongjoon-hyun commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4186037696
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -983,9 +983,10 @@ class PlanMerger(
// 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.
Review 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,
```suggestion
// 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.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala:
##########
@@ -389,6 +390,34 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-59721: ignore an unresolvable ordering reported with
connector-defined classes") {
+ 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 scanRelation = getScanRelation(df)
+ val expectedWarning =
+ s"Spark ignores the ordering reported by ${scanRelation.name} " +
+ 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"))
Review 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?
```suggestion
.filter(_.contains("columns cannot be resolved"))
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3545,6 +3551,160 @@ 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(): (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("columns cannot be resolved")).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(identity("missing"), bucket(4, "absent"), identity("missing")),
+ Some("missing, absent")),
+ (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()
+ 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") {
Review 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,
```suggestion
test("SPARK-59721: a reported ordering with an unresolvable column is
ignored") {
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3545,6 +3551,160 @@ 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(): (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("columns cannot be resolved")).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(identity("missing"), bucket(4, "absent"), identity("missing")),
+ Some("missing, absent")),
+ (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()
+ 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), "id"),
+ (sort(FieldReference("ID"), SortDirection.ASCENDING), "ID"),
+ (sort(FieldReference("index"), SortDirection.ASCENDING), "index"),
Review 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,
```suggestion
// `index` is a metadata column of InMemoryBaseTable.
(sort(FieldReference("index"), SortDirection.ASCENDING), "index"),
```
--
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]
