anuragmantri commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4149110159
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -135,7 +135,8 @@ trait DataSourceV2ScanExecBase
/**
* Returns the output ordering for this scan. When the source reports
ordering via
- * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise,
when the output
+ * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it,
that ordering is
Review Comment:
Done. I updated both the `partitionKeyOrdering.enabled` doc in `SQLConf` and
`docs/sql-performance-tuning.md` with your wording
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2927,9 +2927,10 @@ class MergeSubplansSuite extends PlanTest {
test("SPARK-40259: merge DSv2 scans that report empty key-grouped
partitioning or ordering") {
// A source implementing SupportsReportPartitioning/SupportsReportOrdering
but reporting nothing
- // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the
field unconditionally). An
- // empty report carries no partitioning/ordering to drop, so the merge
should still proceed --
- // the gate tests the inner Seq, not the Option. The fused plan is the
plain column union.
+ // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the
field unless a reported
+ // column cannot be resolved). An empty report carries no
partitioning/ordering to drop, so the
+ // merge should still proceed -- the gate tests the inner Seq, not the
Option. The fused plan is
+ // the plain column union.
Review Comment:
Thanks, applied your wording. The comment now says only the ordering pass
yields `Some(Nil)`, and that the partitioning pass turns an empty report into
`None`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -135,7 +135,8 @@ trait DataSourceV2ScanExecBase
/**
* Returns the output ordering for this scan. When the source reports
ordering via
- * `SupportsReportOrdering`, that ordering is returned as-is. Otherwise,
when the output
+ * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it,
that ordering is
+ * returned as-is. Otherwise, when the output
Review Comment:
Thanks, I reproduced it on both this branch and master. With a columnar
source that reports `KeyGroupedPartitioning(k)` and ordering `[k, a, c]`,
prunes `c`, and has `preserveOrderingOnCoalesce` enabled, a window over
`(PARTITION BY k ORDER BY a)` plans `GroupPartitions(SortedMerge) ->
ColumnarToRow -> BatchScan[k, a]` with no `ProjectExec` in between. The task
fails with `INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND` ("Could not find c#2 in
[k#0,a#1]"). The row-based variant is fine because `ProjectExec` truncates the
ordering.
Trimming `DataSourceV2ScanExecBase.outputOrdering` with
`takeWhile(_.references.subsetOf(outputSet))`, as `doCanonicalize` does, fixes
it and keeps the related suites green. Since it's pre-existing and independent
of this PR, I'll keep it out of here and file a separate JIRA and PR with a
test, unless you already have one in progress.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -46,14 +46,20 @@ import org.apache.spark.util.ArrayImplicits._
object V2ExpressionUtils extends SQLConfHelper with Logging {
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ /**
+ * Variant of `resolveRef` that returns `None` if no attribute matches the
reference. It still
+ * throws for a missing nested field or an ambiguous reference.
+ */
+ private[sql] def resolveRefOpt[T <: NamedExpression](
Review Comment:
Done. `resolveRefOpt` is non-generic and returns `Option[NamedExpression]`,
`resolveRef` keeps the cast, and I removed the type arguments at the call sites
and the `NamedExpression` import in `V2ScanPartitioningAndOrdering`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3486,6 +3492,139 @@ class KeyGroupedPartitioningSuite
}
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+
+ 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 customReportingJoinDf(): DataFrame = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingCatalogName.ns.$table 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") {
+ val nestedTransform =
+ ApplyTransform("f", Seq(ApplyTransform("g",
Seq(FieldReference("missing")))))
+ // 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])](
Review Comment:
Added `SPARK-59721: ignore an unresolvable ordering reported with
connector-defined classes` to `DataSourceV2Suite`, next to the existing
`ordering and partitioning reporting` test. It reads both the Scala and Java
`OrderAndPartitionAwareDataSource` with `partitionKeys=i` and
`orderKeys=missing`, and checks that the query succeeds, the reported ordering
is dropped, and the warning names `missing`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +89,43 @@ 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 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(", "))}.")
+ // With no reported ordering, DataSourceV2ScanExecBase.outputOrdering
may still derive one
+ // from a kept key-grouped partitioning, which does not depend on the
dropped report.
+ d.copy(ordering = None)
+ } else {
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // This is not `V2Expression.references()` because `ApplyTransform` and
`BucketTransform`
+ // override it to return only their top-level arguments, which would miss
`f(g(missing))`.
+ def references(expr: V2Expression): Seq[NamedReference] = expr match {
+ case ref: NamedReference => Seq(ref)
+ case other => other.children().toImmutableArraySeq.flatMap(references)
+ }
+ exprs.flatMap(references).distinct.flatMap { ref =>
+ try {
+ if (V2ExpressionUtils.resolveRefOpt[NamedExpression](ref,
relation).isDefined) {
+ None
+ } else {
+ Some(ref.describe())
Review Comment:
Thanks for the catch. The name is now built from `fieldNames` like
`resolveRef` does, and de-duplicated on the string, so a connector-defined
`NamedReference` prints its column name. The new `DataSourceV2Suite` test
covers it. I checked that it fails when `describe()` is used.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9259,3 +9398,51 @@ 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 {
+ @volatile var reportedKeys: Seq[Expression] = Seq.empty
Review Comment:
Done, removed `@volatile`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2927,9 +2927,10 @@ class MergeSubplansSuite extends PlanTest {
test("SPARK-40259: merge DSv2 scans that report empty key-grouped
partitioning or ordering") {
// A source implementing SupportsReportPartitioning/SupportsReportOrdering
but reporting nothing
- // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the
field unconditionally). An
- // empty report carries no partitioning/ordering to drop, so the merge
should still proceed --
- // the gate tests the inner Seq, not the Option. The fused plan is the
plain column union.
+ // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the
field unless a reported
Review Comment:
Done, I added "unless a reported column cannot be resolved" to the comment
above `mergeDegradesReporting`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +89,43 @@ 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, _) =>
Review Comment:
Thanks, I reproduced it. With a reported ordering on the nested field `s.x`,
the "Early Filter and Projection Push-Down" batch fails its idempotence check
in testing, because each run resolves `s.x` to a new
`Alias(GetStructField(...))`. With that check skipped, the aliased sort order
never satisfies a required ordering on `s.x`, so Spark adds a redundant
`SortExec`. Trimming aliases from the converted ordering fixes both, and the
existing suites still pass.
Since it's pre-existing and changes planning behavior on its own, I'll keep
it out of this PR and file a separate JIRA and PR. I also left out a nested
sort key test here for the same reason.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3486,6 +3492,139 @@ class KeyGroupedPartitioningSuite
}
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+
+ 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 " +
Review Comment:
Done, the `INSERT` and the `SELECT` in `customReportingJoinDf` both use
`customReportingTableName` now.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +89,43 @@ 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 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(", "))}.")
+ // With no reported ordering, DataSourceV2ScanExecBase.outputOrdering
may still derive one
+ // from a kept key-grouped partitioning, which does not depend on the
dropped report.
+ d.copy(ordering = None)
+ } else {
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ // This is not `V2Expression.references()` because `ApplyTransform` and
`BucketTransform`
+ // override it to return only their top-level arguments, which would miss
`f(g(missing))`.
Review Comment:
Updated the comment to your wording. I also widened the `catch` comment and
the `resolveRefOpt` scaladoc to "a nested-field extraction error or an
ambiguous reference", which also covers `AMBIGUOUS_REFERENCE_TO_FIELDS` and
`INVALID_EXTRACT_BASE_FIELD_TYPE`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3486,6 +3492,139 @@ class KeyGroupedPartitioningSuite
}
}
+ private val customReportingCatalogName = "custom_reporting_cat"
+ private val customReportingTableName =
s"$customReportingCatalogName.ns.$table"
+
+ 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 customReportingJoinDf(): DataFrame = sql(
+ s"""SELECT t1.data, t2.data FROM $customReportingCatalogName.ns.$table 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") {
+ val nestedTransform =
+ ApplyTransform("f", Seq(ApplyTransform("g",
Seq(FieldReference("missing")))))
+ // 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(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 (FIELD_NOT_FOUND)")))
+ 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"reported by
$customReportingTableName ")), warnings)
+
assert(warnings.forall(_.contains(classOf[CustomReportingScan].getName)),
warnings)
+ 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)
+ 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"))
+ reportingCatalog.reportedOrdering = Seq(
+ sort(FieldReference("id"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST))
+ val (_, keptScan, keptWarnings) =
+ customReportingJoin("ordering columns cannot be resolved")
+ assert(keptWarnings.isEmpty, keptWarnings)
+
assert(keptScan.ordering.map(_.map(_.child.references.map(_.name).toSeq)) ==
+ Some(Seq(Seq("id"))))
+
+ reportingCatalog.reportedOrdering = Seq(
+ sort(FieldReference("missing"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST))
+ val (_, reportingScan, warnings) =
+ customReportingJoin("ordering columns cannot be resolved")
+ assert(warnings.nonEmpty)
+ assert(warnings.forall(_.contains(s"reported by
$customReportingTableName ")), warnings)
+
assert(warnings.forall(_.contains(classOf[CustomReportingScan].getName)),
warnings)
+ assert(warnings.forall(_.contains("resolved: missing.")), warnings)
+ assert(reportingScan.ordering.isEmpty)
+ assert(reportingScan.keyGroupedPartitioning.isDefined)
+ val derivedOrdering = reportingScan.outputOrdering.map { o =>
+ (o.child.references.map(_.name).toSeq, o.direction)
+ }
+ assert(derivedOrdering == Seq((Seq("id"), Ascending)))
+ }
+ }
+ }
+
+ test("SPARK-59721: a reported partition key that cannot be converted still
fails") {
Review Comment:
Added `[id + 1, identity(missing)]` to the fallback test. It succeeds with a
warning that names only `missing`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -47,9 +49,21 @@ 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 keys = kgp.keys()
+ val unresolvedColumns =
unresolvableColumns(keys.toImmutableArraySeq, relation)
+ val partitioning = if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the KeyGroupedPartitioning reported by " +
+ log"${MDC(RELATION_NAME, relation.name)} (scan " +
+ log"${MDC(CLASS_NAME, scan.getClass.getName)}) because the
partition key columns " +
+ log"cannot be resolved: ${MDC(COLUMN_NAMES,
unresolvedColumns.mkString(", "))}. " +
+ log"Storage-partitioned join will not be applied for this
scan.")
Review Comment:
Removed the sentence, so it matches the ordering warning.
--
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]
