dongjoon-hyun commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4128238826
##########
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:
Now that `V2ScanPartitioningAndOrdering` checks every key reference against
`relation` before calling `toCatalystOpt`, the `None`-on-unresolved path here
is reachable only from tests: `partitioning()` is the only production caller of
`toCatalystOpt`, and `toCatalystTransformOpt` has no production caller. The
only production-visible effect left is that a nested transform whose function
cannot be loaded drops the partitioning instead of failing the query.
Could we keep `V2ExpressionUtils` unchanged except for `resolveRefOpt`? Then
this PR becomes smaller, the contract of
`toCatalystOpt`/`toCatalystTransformOpt` and the nested-transform behavior stay
as they are, and the new `KeyGroupedPartitioningSuite` cases should still pass.
If you prefer to keep it as a defensive change, could you mention it in the PR
description?
##########
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:
Could you also assert `collectShuffles(plan).isEmpty` for this control case?
Otherwise, `collectShuffles(plan).nonEmpty` in the fallback cases would pass
even if this fixture could never plan a storage-partitioned join.
BTW, `CustomReportingCatalog` extends `InMemoryTableCatalog`, which is not a
`FunctionCatalog`, so the `bucket(4, missing)` and `f(g(missing))` keys would
be dropped even if their columns were resolvable. It's fine because the warning
assertion distinguishes them, but a short comment would help.
##########
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:
This warning does not say which relation reports the key, and
`kgp.getClass.getSimpleName` is `KeyGroupedPartitioning` in practice, so it's
hard to find the connector at fault in a query with several V2 scans. Could you
include the relation name like `V2ScanRelationPushDown` does, e.g.,
`${MDC(RELATION_NAME, relation.name)}`, here and in the ordering warning below?
In addition, the reason of the caught `AnalysisException` is dropped, so an
ambiguous column is reported as "cannot be resolved" in the same way as a
missing one.
##########
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:
nit. `scan.outputOrdering()` is now called twice, and `kgp.keys()` above
too. Could you keep the result in a local variable? A connector may build a new
array on each call.
##########
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:
nit. Could you add a short comment on why this does not use
`Expression.references()`? `ApplyTransform` and `BucketTransform` override it
with 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:
FYI, the motivating case in apache/spark#55518 (a relation narrowed for
column-level UPDATE) is not a connector bug, so this warning would be logged
whenever the narrowed relation omits a partition source column if
apache/spark#55518 lands as is. WARN looks fine for this PR because it helps to
find connector bugs, but we may need to revisit the level there.
##########
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:
nit. This file uses `java.util.Map` fully qualified elsewhere (e.g.,
`emptyProps`). Could you use `java.util.Map` in `CustomReportingCatalog` and
remove this import?
##########
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:
Before this PR, a `SupportsReportOrdering` scan always had `ordering =
Some(...)`. With `None` here, `DataSourceV2ScanExecBase.outputOrdering` falls
back to the partition-key-derived ordering when the scan keeps a key-grouped
partitioning and `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
is true (the default), while a scan reporting an empty ordering (`Some(Nil)`)
does not get it. It looks correct to me because the derived ordering does not
depend on the dropped report. Could you add a short comment here and assert
`reportingScan.outputOrdering` in the new ordering test so that this behavior
is explicit?
--
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]
