anuragmantri commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4170238966
##########
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:
Thanks, you're right. Done.
##########
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:
Added your case, `[identity(missing), bucket(4, absent),
identity(missing)]`, which expects missing, absent. I checked that it fails if
unresolvableColumns keeps only the first name.
##########
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:
Removed the withSQLConf.
##########
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:
Done. The test uses getScanRelation(df) and compares the whole warning,
built from scanRelation.name and scanRelation.scan.getClass.getName. It also
compares the set of warnings now, as KeyGroupedPartitioningSuite does.
##########
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:
Removed. `ChildrenOverridingTransform` now keeps the default `references()`,
like `ChildrenOverridingSortOrder`.
##########
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:
Done. `fieldNames` is read once into parts and reused for the empty check
and the quoted name. As in your suggestion, `resolveRefOpt` still reads it once
more inside.
##########
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:
Done. `customReportingJoin()` now uses the fixed filter _.contains("columns
cannot be resolved"), and the parameter is gone.
##########
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:
Done. All the sort calls in this test use the two-argument form.
--
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]
