anuragmantri commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4126805286
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -39,24 +39,38 @@ import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.connector.PartitionPredicateImpl
import org.apache.spark.sql.types._
import org.apache.spark.util.ArrayImplicits._
+import org.apache.spark.util.collection.Utils.sequenceToOption
/**
* A utility class that converts public connector expressions into Catalyst
expressions.
*/
object V2ExpressionUtils extends SQLConfHelper with Logging {
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ /**
+ * Non-throwing variant of `resolveRef`. Returns `None` if the reference
cannot be resolved.
Review Comment:
You're right, thanks. I updated the scaladoc to say it still throws for a
missing nested field or an ambiguous reference. The check in
`V2ScanPartitioningAndOrdering` now treats an `AnalysisException` from
`resolveRefOpt` as unresolvable, so a key like `s.missing` falls back with the
warning instead of failing the query. `resolveRef` is unchanged, so the
throwing APIs keep `FIELD_NOT_FOUND`. The end-to-end test covers `s.missing` on
a struct column.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
+ val unresolvedRefs = kgp.keys().flatMap(_.references()).filter {
ref =>
Review Comment:
Good catch. The check now collects the key references recursively through
`children()`, so a key like `f(g(missing))` is caught. It also runs before
`toCatalystOpt` now, so the partitioning is dropped and the warning is logged
from the same place. I added `f(g(missing))` to the end-to-end test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
+ val unresolvedRefs = kgp.keys().flatMap(_.references()).filter {
ref =>
+ V2ExpressionUtils.resolveRefOpt[NamedExpression](ref,
relation).isEmpty
+ }
+ if (unresolvedRefs.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ${MDC(CLASS_NAME,
kgp.getClass.getSimpleName)} " +
+ log"because the partition key columns cannot be resolved: " +
+ log"${MDC(COLUMN_NAMES,
unresolvedRefs.map(_.describe()).mkString(", "))}. " +
+ log"Storage-partitioned join will not be applied for this
scan.")
Review Comment:
Good point. `ordering()` now runs the same column check on the reported
ordering. If a column can't be resolved, it logs a warning and drops the
reported ordering instead of failing. I added a test where the scan reports an
ordering on `missing`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -9259,3 +9304,43 @@ class KeyGroupedPartitioningCatalystRuntimeFilterSuite
catalog.clearTables()
}
}
+
+/**
+ * An in-memory catalog whose tables report a `KeyGroupedPartitioning` keyed
on a column that
+ * does not exist in the table.
+ */
+class UnresolvablePartitionKeyCatalog extends InMemoryTableCatalog {
+ // scalastyle:off argcount
+ override protected def newInMemoryTable(
+ name: String,
+ columns: Array[Column],
+ partitioning: Array[Transform],
+ properties: 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 =
UnresolvablePartitionKeyScan(super.build())
+ }
+ }
+ }
+}
+
+case class UnresolvablePartitionKeyScan(inner: Scan) extends
SupportsReportPartitioning {
+ override def readSchema(): StructType = inner.readSchema()
+ override def toBatch: Batch = inner.toBatch
+ override def description(): String = inner.description()
+ override def outputPartitioning(): Partitioning = new KeyGroupedPartitioning(
+ Array[Expression](FieldReference("missing")),
Review Comment:
Done. Each test now sets the keys (and ordering) the fixture scan reports,
so one table covers all the cases. The end-to-end test now covers a resolvable
control key, `bucket(4, missing)`, `f(g(missing))`, `[id, missing]`, `[missing,
bucket(4, missing)]`, and `s.missing`. For each unresolvable case it checks
that the warning names exactly the unresolvable columns, that the whole
partitioning is dropped, and that a shuffle is planned. For `[id, missing]`,
only `missing` is listed.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
+ val unresolvedRefs = kgp.keys().flatMap(_.references()).filter {
ref =>
+ V2ExpressionUtils.resolveRefOpt[NamedExpression](ref,
relation).isEmpty
+ }
+ if (unresolvedRefs.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ${MDC(CLASS_NAME,
kgp.getClass.getSimpleName)} " +
+ log"because the partition key columns cannot be resolved: " +
+ log"${MDC(COLUMN_NAMES,
unresolvedRefs.map(_.describe()).mkString(", "))}. " +
Review Comment:
Done, the column names are de-duplicated, and the `[missing, bucket(4,
missing)]` case checks that `missing` is listed once. Agreed on the repeated
logging; I left that as is, like the existing warning.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala:
##########
@@ -20,21 +20,84 @@ package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
+import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryCatalog}
+import org.apache.spark.sql.connector.catalog.functions.{BoundFunction,
ScalarFunction, UnboundFunction}
import org.apache.spark.sql.connector.expressions._
-import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.types.{DataType, IntegerType, StringType,
StructType}
class V2ExpressionUtilsSuite extends SparkFunSuite {
+ private val plan = LocalRelation.apply(AttributeReference("a", StringType)())
+
+ private object AnyInputFunction extends UnboundFunction {
+ override def name(): String = "any_input"
+ override def description(): String = name()
+ override def bind(inputType: StructType): BoundFunction = new
ScalarFunction[Int] {
+ override def inputTypes(): Array[DataType] =
inputType.fields.map(_.dataType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "any_input"
+ }
+ }
+
+ private val funCatalog = {
+ val catalog = new InMemoryCatalog
+ Seq("bucket", "years").foreach { name =>
+ catalog.createFunction(Identifier.of(Array.empty, name),
AnyInputFunction)
+ }
+ Some(catalog)
+ }
+
test("SPARK-39313: toCatalystOrdering should fail if V2Expression can not be
translated") {
val supportedV2Sort = SortValue(
FieldReference("a"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
val unsupportedV2Sort = supportedV2Sort.copy(
expression = ApplyTransform("v2Fun", FieldReference("a") :: Nil))
val exc = intercept[AnalysisException] {
- V2ExpressionUtils.toCatalystOrdering(
- Array(supportedV2Sort, unsupportedV2Sort),
- LocalRelation.apply(AttributeReference("a", StringType)()))
+ V2ExpressionUtils.toCatalystOrdering(Array(supportedV2Sort,
unsupportedV2Sort), plan)
}
assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently
supported"))
}
+
+ test("SPARK-59721: toCatalystOpt returns None, not throw, for an
unresolvable FieldReference") {
+ assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"),
plan).isEmpty)
+ }
+
+ test("SPARK-59721: toCatalyst throws the specific resolution error for an
unresolvable " +
Review Comment:
Added `toCatalyst(Expressions.bucket(4, "missing"), plan, funCatalog)` and
`toCatalyst(Expressions.years("missing"), plan, funCatalog)`, both asserting
`_LEGACY_ERROR_TEMP_1137` with `checkError`. I also switched the existing error
assertions in this suite to `checkError`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -142,27 +166,34 @@ object V2ExpressionUtils extends SQLConfHelper with
Logging {
def toCatalystTransformOpt(
trans: Transform,
query: LogicalPlan,
- funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
trans match {
+ funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
+ toCatalystTransformCore(trans, query, funCatalogOpt, failOnUnresolved =
false)
+
+ private def toCatalystTransformCore(
+ trans: Transform,
+ query: LogicalPlan,
+ funCatalogOpt: Option[FunctionCatalog],
+ failOnUnresolved: Boolean): Option[Expression] = trans match {
case IdentityTransform(ref) =>
- Some(resolveRef[NamedExpression](ref, query))
+ resolveRefOrDegrade[NamedExpression](ref, query, failOnUnresolved)
case BucketTransform(numBuckets, refs, sorted)
if sorted.isEmpty && refs.length == 1 &&
refs.forall(_.isInstanceOf[NamedReference]) =>
- val resolvedRefs = refs.map(r => resolveRef[NamedExpression](r, query))
// Create a dummy reference for `numBuckets` here and use that, together
with `refs`, to
// look up the V2 function.
val numBucketsRef = AttributeReference("numBuckets", IntegerType,
nullable = false)()
- funCatalogOpt.flatMap { catalog =>
- loadV2FunctionOpt(catalog, "bucket", Seq(numBucketsRef) ++
resolvedRefs).map { bound =>
- TransformExpression(bound, resolvedRefs, Some(numBuckets))
- }
- }
+ for {
+ resolvedRefs <- sequenceToOption(
+ refs.map(r => resolveRefOrDegrade[NamedExpression](r, query,
failOnUnresolved)))
+ catalog <- funCatalogOpt
+ bound <- loadV2FunctionOpt(catalog, "bucket", Seq(numBucketsRef) ++
resolvedRefs)
+ } yield TransformExpression(bound, resolvedRefs, Some(numBuckets))
case NamedTransform(name, args) =>
- val catalystArgs = args.map(toCatalyst(_, query, funCatalogOpt))
- funCatalogOpt.flatMap { catalog =>
- loadV2FunctionOpt(catalog, name, catalystArgs).map { bound =>
- TransformExpression(bound, catalystArgs)
- }
- }
+ for {
+ catalystArgs <- sequenceToOption(
+ args.map(toCatalystCore(_, query, funCatalogOpt, failOnUnresolved)))
Review Comment:
Added a test: for `f(g(a))` where `g` is not in the function catalog,
`toCatalystTransformOpt` returns `None`, and `toCatalyst` still fails with
`_LEGACY_ERROR_TEMP_3054` naming `f(g(a))`. The silent drop matches what
already happens for a top-level transform whose function can't be loaded (no
function catalog, or `bind` throwing `UnsupportedOperationException`), so I
kept the two consistent.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -50,6 +50,18 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
val partitioning = sequenceToOption(
kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation,
relation.funCatalog))
.toImmutableArraySeq)
+ if (partitioning.isEmpty) {
Review Comment:
Yes, I changed it that way. The rule now checks the key references
recursively against `relation` first. If any can't be resolved, it logs the
warning and uses `None` without calling `toCatalystOpt`; otherwise the existing
path runs unchanged. The reported ordering uses the same check.
--
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]
