anuragmantri commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4140798412
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3396,6 +3403,97 @@ class KeyGroupedPartitioningSuite
}
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+
+ 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 $customReportingCatalogName.ns.$table 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 customReportingJoin(warningText: String): (SparkPlan,
BatchScanExec, Seq[String]) = {
+ val df = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingCatalogName.ns.$table t1
+ |JOIN testcat.ns.$table t2 ON t1.id = t2.id
+ |""".stripMargin)
+ 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")))))
+ 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(identity("s.missing")), Some("s.missing")))
+ 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"resolved: $columnsString.")),
warnings)
+ assert(reportingScan.keyGroupedPartitioning.isEmpty, keysString)
+ assert(collectShuffles(plan).nonEmpty, keysString)
+ case None =>
+ assert(warnings.isEmpty, warnings)
+ assert(reportingScan.keyGroupedPartitioning.isDefined, keysString)
Review Comment:
Done. The control case now asserts `collectShuffles(plan).isEmpty`, which
confirms the fixture does plan a storage-partitioned join when the key
resolves. I also added a comment that `CustomReportingCatalog` is not a
`FunctionCatalog`, and that the warning assertion is what shows the
unresolvable column causes the drop.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +87,34 @@ 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, _) =>
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val unresolvedColumns =
+ unresolvableColumns(scan.outputOrdering().toImmutableArraySeq,
relation)
Review Comment:
Done, `kgp.keys()` and `scan.outputOrdering()` are each read once into a
local variable now.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -47,9 +49,19 @@ 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 unresolvedColumns =
unresolvableColumns(kgp.keys().toImmutableArraySeq, relation)
+ val partitioning = if (unresolvedColumns.nonEmpty) {
+ logWarning(
Review Comment:
Done. Both warnings now include `${MDC(RELATION_NAME, relation.name)}` and I
replaced `kgp.getClass.getSimpleName` with the scan class, so the connector at
fault is easy to find. When `resolveRefOpt` throws, the column is listed with
the error condition, e.g. `s.missing (FIELD_NOT_FOUND)` or `a
(AMBIGUOUS_REFERENCE)`. The end-to-end tests check the relation name and the
`s.missing (FIELD_NOT_FOUND)` entry.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +87,34 @@ 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, _) =>
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val unresolvedColumns =
+ unresolvableColumns(scan.outputOrdering().toImmutableArraySeq,
relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ordering because the ordering columns
cannot be " +
+ log"resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(",
"))}.")
+ d.copy(ordering = None)
Review Comment:
Agreed. I added a comment that with no reported ordering,
`DataSourceV2ScanExecBase.outputOrdering` may still derive one from a kept
key-grouped partitioning, and that this doesn't depend on the dropped report. I
also updated the `outputOrdering` scaladoc, which said a reported ordering is
always returned as-is. The ordering test now sets
`partitionKeyOrdering.enabled` explicitly, asserts that
`reportingScan.outputOrdering` is the derived `id ASC`, and has a control case
where a resolvable reported ordering is kept without a warning.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -113,25 +128,35 @@ object V2ExpressionUtils extends SQLConfHelper with
Logging {
expr: V2Expression,
query: LogicalPlan,
funCatalogOpt: Option[FunctionCatalog] = None): Expression =
- toCatalystOpt(expr, query, funCatalogOpt)
+ toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = true)
.getOrElse(throw new AnalysisException(
errorClass = "_LEGACY_ERROR_TEMP_3054", messageParameters =
Map("expr" -> expr.toString)))
def toCatalystOpt(
expr: V2Expression,
query: LogicalPlan,
- funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] = {
+ funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
+ toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = false)
Review Comment:
Following Dongjoon's suggestion, I reverted the `V2ExpressionUtils` changes,
so `toCatalystOpt` and `toCatalystTransformOpt` keep their original contract
and this PR no longer changes them.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -17,8 +17,11 @@
package org.apache.spark.sql.connector
import java.sql.Timestamp
+import java.util
Review Comment:
Done, the fixture now uses `java.util.Map` and the import is removed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -113,25 +128,35 @@ object V2ExpressionUtils extends SQLConfHelper with
Logging {
expr: V2Expression,
query: LogicalPlan,
funCatalogOpt: Option[FunctionCatalog] = None): Expression =
- toCatalystOpt(expr, query, funCatalogOpt)
+ toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = true)
.getOrElse(throw new AnalysisException(
errorClass = "_LEGACY_ERROR_TEMP_3054", messageParameters =
Map("expr" -> expr.toString)))
def toCatalystOpt(
expr: V2Expression,
query: LogicalPlan,
- funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] = {
+ funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
+ toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = false)
+
+ /**
+ * Shared implementation of `toCatalyst`/`toCatalystOpt`, keyed on
`failOnUnresolved`.
+ */
+ private def toCatalystCore(
Review Comment:
Agreed, I reverted it. `V2ExpressionUtils` is now unchanged except for
adding `resolveRefOpt`, with `resolveRef` as a thin wrapper around it. I
removed the `V2ExpressionUtilsSuite` tests that were tied to the refactor and
kept tests for `resolveRefOpt`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +87,34 @@ 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, _) =>
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val unresolvedColumns =
+ unresolvableColumns(scan.outputOrdering().toImmutableArraySeq,
relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ordering because the ordering columns
cannot be " +
+ log"resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(",
"))}.")
+ d.copy(ordering = None)
+ } else {
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(),
relation, relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ def references(expr: V2Expression): Seq[NamedReference] = expr match {
Review Comment:
Done. I added a comment that `ApplyTransform` and `BucketTransform` override
`references()` to return only their top-level arguments, so it would miss
`f(g(missing))`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -47,9 +49,19 @@ 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 unresolvedColumns =
unresolvableColumns(kgp.keys().toImmutableArraySeq, relation)
+ val partitioning = if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ${MDC(CLASS_NAME,
kgp.getClass.getSimpleName)} " +
+ log"because the partition key columns cannot be resolved: " +
Review Comment:
Thanks for the heads-up. Agreed that `WARN` fits here, since an unresolvable
key usually means a connector bug. I'll revisit the level in
apache/spark#55518, where a narrowed relation can legitimately omit a partition
source column.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +87,34 @@ 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, _) =>
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val unresolvedColumns =
+ unresolvableColumns(scan.outputOrdering().toImmutableArraySeq,
relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ordering because the ordering columns
cannot be " +
+ log"resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(",
"))}.")
+ d.copy(ordering = None)
+ } else {
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(),
relation, relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ def references(expr: V2Expression): Seq[NamedReference] = expr match {
+ case ref: NamedReference => Seq(ref)
+ case other => other.children().toImmutableArraySeq.flatMap(references)
+ }
+ exprs.flatMap(references).distinct.filter { ref =>
+ try {
+ V2ExpressionUtils.resolveRefOpt[NamedExpression](ref, relation).isEmpty
+ } catch {
+ // A missing nested field or an ambiguous name throws instead of
returning None.
+ case _: AnalysisException => true
Review Comment:
Added your test to `V2ExpressionUtilsSuite`, using `checkError` with
`AMBIGUOUS_REFERENCE`, together with the `FIELD_NOT_FOUND` case for
`s.missing`. Both are the preconditions this `catch` relies on.
--
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]
