anuragmantri commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4186260135
##########
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:
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".
##########
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:
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.
##########
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:
Done. The filter is now `_.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") {
+ 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:
Added the comment.
--
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]
