spark-test commented on code in PR #59909: URL: https://github.com/apache/spark/pull/59909#discussion_r4418015079
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExternalLookupJoin.scala: ########## @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.catalyst.analysis + +import org.apache.spark.sql.catalyst.optimizer.EliminateResolvedHint +import org.apache.spark.sql.catalyst.plans.{Inner, LeftOuter} +import org.apache.spark.sql.catalyst.plans.logical.{ExternalLookupJoin, Join, JoinHint, + JoinStrategyHint, LogicalPlan} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.JOIN +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase} +import org.apache.spark.sql.execution.datasources.v2.ExternalLookupRelations + +/** + * Replaces an eligible join with a lookup join holder. + * + * A join is planned as an external lookup join when its right side reads a table that supports + * external lookup but not batch read. It must be a stream-static join: the left side is a + * streaming query, whose micro-batches tolerate lookup results that change between requests, and + * the lookup-only table is on the right side. + */ +object ResolveExternalLookupJoin extends Rule[LogicalPlan] with QueryErrorsBase { + private val SupportedJoinTypes = Seq(Inner, LeftOuter) + + override def apply(plan: LogicalPlan): LogicalPlan = + plan.resolveOperatorsUpWithPruning(_.containsPattern(JOIN), ruleId) { + case join: Join if join.left.resolved && join.right.resolved => resolveLookupJoin(join) Review Comment: **Non-blocking (P2):** This guard checks only that both children are resolved. `Join.resolved` also requires `duplicateResolved` and a `BooleanType` condition, but `ExternalLookupJoin` inherits the default `resolved`, and the checks that enforce those invariants match only `Join`: `JOIN_CONDITION_IS_NOT_BOOLEAN_TYPE` and the conflicting-attributes check in CheckAnalysis, and the right-side dedup in `DeduplicateRelations`. `DeduplicateRelations` also skips streaming subtrees in its renewal pass, and every lookup join has a streaming left side. Once the rule is registered, `stream.join(lookupDf, stream("id"))`, whose condition is INT, passes analysis, keeps the INT expression as the whole residual, and fails in preparation with `NO_LOOKUP_PREDICATE`, which blames the data source. For chained lookups against the same relation instance, the conversion can fire before `DeduplicateRelations` renews the right side, depending only on where the rule lands in the Resolution batch, and the conflicting exprIds are then never repaired or reported. Matching on `join.resolved` lets the Join owners finish first. This matters only once SPARK-60690 places the rule in the analyzer. **Verification:** - **Regression:** Applying the rule to a join with an INT condition over a lookup-only table leaves the join unchanged, so analysis reports that the join condition is not boolean. - **Regression:** Applying the rule to a join whose right side shares attribute IDs with its left side leaves the join unchanged until the right side is deduplicated. - **Compatibility:** Resolved inner and left outer joins over lookup-only tables still convert, and the existing rejection reasons still fire with the same parameters. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PrepareExternalLookupJoin.scala: ########## @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.datasources.v2 + +import java.util.{Collections, IdentityHashMap} + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualTo, Expression, + PredicateHelper} +import org.apache.spark.sql.catalyst.plans.logical.{ExternalLookupJoin, Filter, Join, LogicalPlan, + PreparedExternalLookupJoin, Project} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.EXTERNAL_LOOKUP_JOIN +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, FieldReference} +import org.apache.spark.sql.connector.expressions.filter.{Predicate => V2Predicate} +import org.apache.spark.sql.connector.read.{LookupParameter, SupportsExternalLookup} +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.DataType + +/** Finalizes an optimized external lookup side immediately before ordinary V2 scan preparation. */ +object PrepareExternalLookupJoin extends Rule[LogicalPlan] with PredicateHelper + with QueryErrorsBase { + override def apply(plan: LogicalPlan): LogicalPlan = { + val prepared = plan.transformUpWithPruning(_.containsPattern(EXTERNAL_LOOKUP_JOIN)) { + case join: ExternalLookupJoin => prepare(join) + } + // Prepared lookup joins no longer contain their lookup relation, so a lookup-only table that + // is still read here cannot run. Pre-order visits a join before the relation it reads. + prepared.foreach { + // Classic Dataset.join(right, usingColumns) first analyzes Join(left, right, joinType, None), + // including CheckAnalysis, to disambiguate a self-join, and only then builds the USING join. + // That plan cannot be told apart from a join without a condition, so analysis keeps it, and + // a missing condition is reported here, on the final plan. + case Join(_, right, _, None, _) if readsLookupOnlyTable(right) => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("MISSING_JOIN_CONDITION") + case node => + ExternalLookupRelations.lookupOnlyTable(node).foreach { table => + throw QueryCompilationErrors.unsupportedBatchReadError(table) + } + } + prepared + } + + private def readsLookupOnlyTable(plan: LogicalPlan): Boolean = plan match { + case node @ (_: Project | _: Filter) => readsLookupOnlyTable(node.children.head) + case node => ExternalLookupRelations.lookupOnlyTable(node).isDefined + } + + private def prepare(join: ExternalLookupJoin): PreparedExternalLookupJoin = { + val relation = extractLookupPlan(join.right).getOrElse { + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("LOOKUP_PLAN") + } + val tableName = relation.table.name() + val supportsRead = relation.table match { + case table: SupportsRead + if table.capabilities().contains(TableCapability.EXTERNAL_LOOKUP) => table + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "TABLE", Map("table" -> toSQLId(tableName))) + } + val scanBuilder = supportsRead.newScanBuilder(relation.options) + val lookupBuilder = scanBuilder match { + case builder: SupportsExternalLookup => builder + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "SCAN_BUILDER", Map("table" -> toSQLId(tableName))) + } + + val conjuncts = join.condition.toSeq.flatMap(splitConjunctivePredicates) + val offeredCandidates = mutable.ArrayBuffer.empty[OfferedLookupPredicate] + val residual = mutable.ArrayBuffer.empty[Expression] + conjuncts.foreach { + case equality @ EqualTo(leftExpr, rightAttr: Attribute) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case equality @ EqualTo(rightAttr: Attribute, leftExpr) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case other => residual += other + } + + val offeredPredicates = offeredCandidates.map { candidate => + new V2Predicate(candidate.predicateName, Array[V2Expression]( + FieldReference.column(candidate.projectedField.name), + new LookupParameter(candidate.parameter.dataType))) + }.toArray + // A copy, so a connector that reorders the array in place cannot break the identity mapping. + val (selectedPredicates, unselectedPredicates) = selectedCandidates(tableName, + offeredPredicates, offeredCandidates, + lookupBuilder.selectLookupPredicates(offeredPredicates.clone())) + residual ++= unselectedPredicates.map(_.condition) + + val scan = scanBuilder.build() + val resolver = SQLConf.get.resolver + // Fields match the way ordinary V2 scans match them (`PushDownUtils.toOutputAttrs`): CHAR and + // VARCHAR are read as STRING, and nullability, including nested nullability, does not count. + val readSchema = CharVarcharUtils.replaceCharVarcharWithStringInSchema(scan.readSchema()) + val lookupInput = readSchema.fields.map { field => + relation.output.find { attr => + resolver(attr.name, field.name) && Review Comment: **Nit (P3):** Under the default `spark.sql.caseSensitive=false`, `resolver(attr.name, field.name)` binds a read-schema field to the first attribute whose name matches case-insensitively, and `DUPLICATE_SCAN_FIELD` then groups the bound attributes by exact name. For a lookup-only table `(k INT, a INT, A INT)`, which `DataSourceV2Relation.create` keeps as three distinct attributes, a full read schema binds `A` to `a`, and the rule rejects a connector that listed each column once with `DUPLICATE_SCAN_FIELD`. With a read schema `(k, A)`, which the contract admits, `A`'s values are bound to attribute `a`, so expressions over `a` read `A`'s values once the join executes. The comment above says matching works like `PushDownUtils.toOutputAttrs`, but that function looks names up exactly, and `Scan.toExternalLookup` asks for the table's name. See **Shared repair plan 1** in the review body. ########## common/utils/src/main/resources/error/error-conditions.json: ########## @@ -2817,6 +2861,59 @@ ], "sqlState" : "42822" }, + "EXTERNAL_LOOKUP_JOIN_UNSUPPORTED" : { + "message" : [ + "The join cannot be executed as an external lookup join:" + ], + "subClass" : { + "CONFLICTING_JOIN_STRATEGY" : { + "message" : [ + "The join strategy hint <strategy> cannot be honored by an external lookup join over <relation>. Remove the <strategy> hint, or the `EXTERNAL_LOOKUP` hint if the table also supports batch reads." Review Comment: **Nit (P3):** No `EXTERNAL_LOOKUP` hint exists yet (the TODO in `ResolveExternalLookupJoin` defers it), and Spark drops unrecognized hints with only a warning, so this remedy cannot be followed. The error is also raised only for lookup-only tables, since `resolveLookupJoin` returns earlier for every other table, so the "if the table also supports batch reads" branch never applies. `LOOKUP_PLAN` below has the same problem: it says the lookup side may be followed by projections and filters, but `extractLookupPlan` accepts only a bare `DataSourceV2Relation` or `ExternalLookupRelation`, and looking through such nodes is a TODO. Stating only the current remedy and form, and extending the texts when those features land, avoids pointing users at options they cannot use. See **Shared repair plan 2** in the review body. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PrepareExternalLookupJoin.scala: ########## @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.datasources.v2 + +import java.util.{Collections, IdentityHashMap} + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualTo, Expression, + PredicateHelper} +import org.apache.spark.sql.catalyst.plans.logical.{ExternalLookupJoin, Filter, Join, LogicalPlan, + PreparedExternalLookupJoin, Project} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.EXTERNAL_LOOKUP_JOIN +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, FieldReference} +import org.apache.spark.sql.connector.expressions.filter.{Predicate => V2Predicate} +import org.apache.spark.sql.connector.read.{LookupParameter, SupportsExternalLookup} +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.DataType + +/** Finalizes an optimized external lookup side immediately before ordinary V2 scan preparation. */ +object PrepareExternalLookupJoin extends Rule[LogicalPlan] with PredicateHelper + with QueryErrorsBase { + override def apply(plan: LogicalPlan): LogicalPlan = { + val prepared = plan.transformUpWithPruning(_.containsPattern(EXTERNAL_LOOKUP_JOIN)) { + case join: ExternalLookupJoin => prepare(join) + } + // Prepared lookup joins no longer contain their lookup relation, so a lookup-only table that + // is still read here cannot run. Pre-order visits a join before the relation it reads. + prepared.foreach { + // Classic Dataset.join(right, usingColumns) first analyzes Join(left, right, joinType, None), + // including CheckAnalysis, to disambiguate a self-join, and only then builds the USING join. + // That plan cannot be told apart from a join without a condition, so analysis keeps it, and + // a missing condition is reported here, on the final plan. + case Join(_, right, _, None, _) if readsLookupOnlyTable(right) => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("MISSING_JOIN_CONDITION") + case node => + ExternalLookupRelations.lookupOnlyTable(node).foreach { table => + throw QueryCompilationErrors.unsupportedBatchReadError(table) + } + } + prepared + } + + private def readsLookupOnlyTable(plan: LogicalPlan): Boolean = plan match { + case node @ (_: Project | _: Filter) => readsLookupOnlyTable(node.children.head) + case node => ExternalLookupRelations.lookupOnlyTable(node).isDefined + } + + private def prepare(join: ExternalLookupJoin): PreparedExternalLookupJoin = { + val relation = extractLookupPlan(join.right).getOrElse { + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("LOOKUP_PLAN") + } + val tableName = relation.table.name() + val supportsRead = relation.table match { + case table: SupportsRead + if table.capabilities().contains(TableCapability.EXTERNAL_LOOKUP) => table + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "TABLE", Map("table" -> toSQLId(tableName))) + } + val scanBuilder = supportsRead.newScanBuilder(relation.options) + val lookupBuilder = scanBuilder match { + case builder: SupportsExternalLookup => builder + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "SCAN_BUILDER", Map("table" -> toSQLId(tableName))) + } + + val conjuncts = join.condition.toSeq.flatMap(splitConjunctivePredicates) + val offeredCandidates = mutable.ArrayBuffer.empty[OfferedLookupPredicate] + val residual = mutable.ArrayBuffer.empty[Expression] + conjuncts.foreach { + case equality @ EqualTo(leftExpr, rightAttr: Attribute) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case equality @ EqualTo(rightAttr: Attribute, leftExpr) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case other => residual += other + } + + val offeredPredicates = offeredCandidates.map { candidate => + new V2Predicate(candidate.predicateName, Array[V2Expression]( + FieldReference.column(candidate.projectedField.name), + new LookupParameter(candidate.parameter.dataType))) + }.toArray + // A copy, so a connector that reorders the array in place cannot break the identity mapping. + val (selectedPredicates, unselectedPredicates) = selectedCandidates(tableName, + offeredPredicates, offeredCandidates, + lookupBuilder.selectLookupPredicates(offeredPredicates.clone())) + residual ++= unselectedPredicates.map(_.condition) Review Comment: **Nit (P3):** Appending the unselected candidates here puts them after every non-candidate conjunct. For `ON t.id = s.id AND t.region = s.region AND assert_true(s.x > 0) IS NULL`, a connector that selects only `id = ?`, which it may since it picks one access path, yields the residual `assert_true(s.x > 0) IS NULL AND t.region = s.region`. The assert then runs on every id-matched row, including rows that `t.region = s.region`, written before it, would have filtered out, and the query can fail with `USER_RAISED_EXCEPTION`. SPARK-60602 restored the rule that a throwable conjunct is not evaluated ahead of conjuncts written before it, and `PushPredicateThroughJoin` keeps throwable conditions out of its pushdown split for the same reason. Rebuilding the residual by walking the conjuncts in written order and dropping only the selected candidates keeps that order. This matters once lookup joins execute. **Verification:** - **Regression:** Take a condition with two lookup equalities followed by an assert_true conjunct, where the connector selects only the first equality. The residual evaluates the second equality before the assert_true conjunct. - **Compatibility:** A condition whose candidates are all selected, or that contains no candidates, produces the same residual as before. ########## common/utils/src/main/resources/error/error-conditions.json: ########## @@ -2350,6 +2350,50 @@ ], "sqlState" : "KD010" }, + "DATA_SOURCE_EXTERNAL_LOOKUP_NOT_SUPPORTED" : { + "message" : [ + "<description>: External lookup is not supported." + ], + "sqlState" : "0A000" + }, + "DATA_SOURCE_INVALID_EXTERNAL_LOOKUP" : { + "message" : [ + "Data source table <table> violates the external lookup contract:" + ], + "subClass" : { + "DUPLICATE_SCAN_FIELD" : { + "message" : [ + "The lookup scan returned the field <field> more than once." + ] + }, + "DUPLICATE_SELECTED_PREDICATE" : { + "message" : [ + "`selectLookupPredicates()` returned the predicate <predicate> more than once." + ] + }, + "NON_POSITIVE_MAX_PARAMETER_ROWS" : { + "message" : [ + "`maxParameterRowsPerRequest()` returned <value>, but it must be positive." + ] + }, + "NULL_PREDICATE_SELECTION" : { + "message" : [ + "`selectLookupPredicates()` returned null or an array containing null." + ] + }, + "UNKNOWN_SCAN_FIELD" : { + "message" : [ + "The lookup scan returned the field <field>, which is not in the table output." Review Comment: **Nit (P3):** `PrepareExternalLookupJoin` raises this when no relation attribute matches a read-schema field by both name and type, and the PR description routes a type mismatch such as `name INT` for a STRING column here. The rule fills `<field>` with the name alone, so a read schema `(id INT, value INT)` for a table `(id INT, value STRING NOT NULL)` produces "The lookup scan returned the field `value`, which is not in the table output", although `value` is in the table output. Mismatches such as `TIMESTAMP_NTZ` for `TIMESTAMP`, or `STRING COLLATE UTF8_LCASE` for a UTF8_BINARY column, are the hardest to spot from that text. Saying that the field matches no table column by name and data type, or reporting the expected and actual types in a separate subclass, keeps the PR's choice of failing these schemas. See **Shared repair plan 2** in the review body. ########## sql/core/src/test/scala/org/apache/spark/sql/connector/PrepareExternalLookupJoinSuite.scala: ########## @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector + +import org.apache.spark.SparkException +import org.apache.spark.sql.AnalysisException +import org.apache.spark.sql.QueryTest +import org.apache.spark.sql.catalyst.analysis.ResolveExternalLookupJoin +import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, EqualTo, Literal} +import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftOuter} +import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, LocalRelation, + PreparedExternalLookupJoin, Project} +import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability} +import org.apache.spark.sql.connector.catalog.TableCapability.EXTERNAL_LOOKUP +import org.apache.spark.sql.connector.expressions.FieldReference +import org.apache.spark.sql.connector.expressions.filter.Predicate +import org.apache.spark.sql.connector.read.{ExternalLookup, ExternalLookupReaderFactory, Scan, + ScanBuilder, SupportsExternalLookup} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, + PrepareExternalLookupJoin} +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.{IntegerType, StructType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +class PrepareExternalLookupJoinSuite extends QueryTest with SharedSparkSession { Review Comment: **Nit (P3):** None of the three tests uses `spark`, SQL or `checkAnswer`: they apply `ResolveExternalLookupJoin` and `PrepareExternalLookupJoin` to hand-built plans and need only `SQLConf.get` plus `checkError` and `intercept`, which `SparkFunSuite` provides. Extending `PlanTest`, which other sql/core suites already use, avoids starting a shared session and follows the repository's guidance to use the lowest base class that provides what a test needs. **Verification:** - **Compile:** The suite compiles against the lighter base class with its existing assertions. - **Behavior:** Every existing preparation test still passes without a shared session. ########## sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveExternalLookupJoinSuite.scala: ########## @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.catalyst.analysis + +import scala.jdk.CollectionConverters._ + +import org.apache.spark.SparkException +import org.apache.spark.sql.AnalysisException +import org.apache.spark.sql.catalyst.dsl.expressions._ +import org.apache.spark.sql.catalyst.plans.{Cross, FullOuter, Inner, LeftOuter, RightOuter} +import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, ExternalLookupJoin, HintInfo, Join, + JoinHint, JoinStrategyHint, LocalRelation, LogicalPlan, SHUFFLE_HASH, SHUFFLE_MERGE, + SHUFFLE_REPLICATE_NL} +import org.apache.spark.sql.connector.catalog.{Table, TableCapability} +import org.apache.spark.sql.connector.catalog.TableCapability.{BATCH_READ, EXTERNAL_LOOKUP} +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +class ResolveExternalLookupJoinSuite extends AnalysisTest { + // The stream side of a stream-static join. + private val left = LocalRelation(Seq($"tenant".int, $"id".int), isStreaming = true) + private val lookupOnly = relation(EXTERNAL_LOOKUP) + + private def relation(capabilities: TableCapability*): DataSourceV2Relation = + DataSourceV2Relation.create( Review Comment: **Nit (P3):** This bare `DataSourceV2Relation` over a table without `BATCH_READ` is the lookup side in every test of both suites, but `TableCapabilityCheck` rejects that relation in any analyzed plan, so once the rules are wired in, the lookup side has to be the `ExternalLookupRelation` placeholder. No test constructs it, so its classification, its unwrapping in `extractLookupPlan`, `newInstance` and the `computeStats` override are untested. The hint tests set `JoinHint.leftHint` directly, while during analysis a strategy hint is a `ResolvedHint` node that only `EliminateResolvedHint` moves into `Join.hint`, in the optimizer, so the realistic path through `extractHintsFromPlan` and the nested-join exclusion are never exercised. No test runs `ColumnPruning` over either node. Removing `requiredLookupOutput` from `ExternalLookupJoin`, or dropping `extractHintsFromPlan` from `joinStrategyHint`, would leave every test green. **Verification:** - **Regression:** Column pruning over a projection above an external lookup join never inserts a projection between the join and its lookup relation, and preparation still succeeds. - **Behavior:** A strategy hint resolved above the streaming side raises CONFLICTING_JOIN_STRATEGY, while a strategy hint inside a nested join on the left does not. - **Behavior:** A lookup join whose right side is an ExternalLookupRelation is converted and prepared, and a stray ExternalLookupRelation outside a lookup join is rejected. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PrepareExternalLookupJoin.scala: ########## @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.datasources.v2 + +import java.util.{Collections, IdentityHashMap} + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualTo, Expression, + PredicateHelper} +import org.apache.spark.sql.catalyst.plans.logical.{ExternalLookupJoin, Filter, Join, LogicalPlan, + PreparedExternalLookupJoin, Project} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.EXTERNAL_LOOKUP_JOIN +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, FieldReference} +import org.apache.spark.sql.connector.expressions.filter.{Predicate => V2Predicate} +import org.apache.spark.sql.connector.read.{LookupParameter, SupportsExternalLookup} +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.DataType + +/** Finalizes an optimized external lookup side immediately before ordinary V2 scan preparation. */ +object PrepareExternalLookupJoin extends Rule[LogicalPlan] with PredicateHelper + with QueryErrorsBase { + override def apply(plan: LogicalPlan): LogicalPlan = { + val prepared = plan.transformUpWithPruning(_.containsPattern(EXTERNAL_LOOKUP_JOIN)) { + case join: ExternalLookupJoin => prepare(join) + } + // Prepared lookup joins no longer contain their lookup relation, so a lookup-only table that + // is still read here cannot run. Pre-order visits a join before the relation it reads. + prepared.foreach { + // Classic Dataset.join(right, usingColumns) first analyzes Join(left, right, joinType, None), + // including CheckAnalysis, to disambiguate a self-join, and only then builds the USING join. + // That plan cannot be told apart from a join without a condition, so analysis keeps it, and + // a missing condition is reported here, on the final plan. + case Join(_, right, _, None, _) if readsLookupOnlyTable(right) => Review Comment: **Nit (P3):** This case only matches a join whose condition is still `None` on the final plan. For `FROM s, t WHERE s.id = t.id` or `s.join(t).where(s("id") === t("id"))`, analysis keeps `Filter(s.id = t.id, Join(s, t, Inner, None))`, because `ResolveExternalLookupJoin` returns conditionless joins unchanged. `PushPredicateThroughJoin`, which runs in operator optimization before the early scan push-down batch, then moves the non-throwable equality into the inner join's condition. The final plan holds `Join(s, t, Inner, Some(s.id = t.id))`, this case does not match, and the lookup-only relation falls through to `unsupportedBatchReadError`. The user is told the table does not support batch scan rather than that the lookup predicate belongs in `ON`, although the PR states that such joins fail with a lookup-join reason, not a generic batch-read error. This applies once the rules are registered. **Recommended change:** Move the missing-join-condition detection out of the final validation pass of the lookup join preparation rule and into a check that runs at the start of optimization, after analysis and before operator optimization. A join over a lookup-only table with no ON condition then fails with MISSING_JOIN_CONDITION even when its WHERE clause would later be pushed into it. Keep the final pass for lookup-only relations that are still read outside a lookup join. **Why this works:** At the start of optimization, analysis has completed, including the Dataset API's analysis of USING joins, and no rule has yet moved WHERE predicates into the user's join. Reporting there neither breaks USING joins nor depends on how operator optimization later rewrites the join. **Scope:** Report missing join conditions over lookup-only tables before operator optimization, and document where the check is meant to run for the registration follow-up. **Risks:** A check that runs before operator optimization also sees conditionless joins that later rewrites remove, so it must apply only to joins that read a lookup-only table directly or through projections and filters. **Constraints:** Do not raise the error during analysis, where the Dataset API's USING-join path analyzes a join without a condition. Leave the accepted batch-scan failure for lookup sides beneath aliases, hints, projections or filters unchanged. **Exclusions:** - Planning WHERE-clause equalities as lookup predicates, which would be a new capability outside this PR's declared scope. **Success:** A stream-static inner join over a lookup-only table whose equality is written in WHERE fails with EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.MISSING_JOIN_CONDITION. A join whose condition is in ON over a direct lookup-only right side is unaffected, and joins whose lookup side is beneath an alias still fail as plain batch scans. **Verification:** - **Regression:** When the check runs before predicate pushdown, a filter over a conditionless inner join with a lookup-only right side fails with MISSING_JOIN_CONDITION instead of a batch-scan error. - **Compatibility:** The USING-join form still analyzes and becomes a lookup join, and a join with an ON condition still prepares as before. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PrepareExternalLookupJoin.scala: ########## @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.datasources.v2 + +import java.util.{Collections, IdentityHashMap} + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualTo, Expression, + PredicateHelper} +import org.apache.spark.sql.catalyst.plans.logical.{ExternalLookupJoin, Filter, Join, LogicalPlan, + PreparedExternalLookupJoin, Project} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.EXTERNAL_LOOKUP_JOIN +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, FieldReference} +import org.apache.spark.sql.connector.expressions.filter.{Predicate => V2Predicate} +import org.apache.spark.sql.connector.read.{LookupParameter, SupportsExternalLookup} +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.DataType + +/** Finalizes an optimized external lookup side immediately before ordinary V2 scan preparation. */ +object PrepareExternalLookupJoin extends Rule[LogicalPlan] with PredicateHelper + with QueryErrorsBase { + override def apply(plan: LogicalPlan): LogicalPlan = { + val prepared = plan.transformUpWithPruning(_.containsPattern(EXTERNAL_LOOKUP_JOIN)) { + case join: ExternalLookupJoin => prepare(join) + } + // Prepared lookup joins no longer contain their lookup relation, so a lookup-only table that + // is still read here cannot run. Pre-order visits a join before the relation it reads. + prepared.foreach { + // Classic Dataset.join(right, usingColumns) first analyzes Join(left, right, joinType, None), + // including CheckAnalysis, to disambiguate a self-join, and only then builds the USING join. + // That plan cannot be told apart from a join without a condition, so analysis keeps it, and + // a missing condition is reported here, on the final plan. + case Join(_, right, _, None, _) if readsLookupOnlyTable(right) => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("MISSING_JOIN_CONDITION") + case node => + ExternalLookupRelations.lookupOnlyTable(node).foreach { table => + throw QueryCompilationErrors.unsupportedBatchReadError(table) + } + } + prepared + } + + private def readsLookupOnlyTable(plan: LogicalPlan): Boolean = plan match { + case node @ (_: Project | _: Filter) => readsLookupOnlyTable(node.children.head) + case node => ExternalLookupRelations.lookupOnlyTable(node).isDefined + } + + private def prepare(join: ExternalLookupJoin): PreparedExternalLookupJoin = { + val relation = extractLookupPlan(join.right).getOrElse { + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("LOOKUP_PLAN") + } + val tableName = relation.table.name() + val supportsRead = relation.table match { + case table: SupportsRead + if table.capabilities().contains(TableCapability.EXTERNAL_LOOKUP) => table + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "TABLE", Map("table" -> toSQLId(tableName))) + } + val scanBuilder = supportsRead.newScanBuilder(relation.options) + val lookupBuilder = scanBuilder match { + case builder: SupportsExternalLookup => builder + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "SCAN_BUILDER", Map("table" -> toSQLId(tableName))) + } + + val conjuncts = join.condition.toSeq.flatMap(splitConjunctivePredicates) + val offeredCandidates = mutable.ArrayBuffer.empty[OfferedLookupPredicate] + val residual = mutable.ArrayBuffer.empty[Expression] + conjuncts.foreach { + case equality @ EqualTo(leftExpr, rightAttr: Attribute) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case equality @ EqualTo(rightAttr: Attribute, leftExpr) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case other => residual += other + } + + val offeredPredicates = offeredCandidates.map { candidate => + new V2Predicate(candidate.predicateName, Array[V2Expression]( + FieldReference.column(candidate.projectedField.name), + new LookupParameter(candidate.parameter.dataType))) + }.toArray + // A copy, so a connector that reorders the array in place cannot break the identity mapping. + val (selectedPredicates, unselectedPredicates) = selectedCandidates(tableName, + offeredPredicates, offeredCandidates, + lookupBuilder.selectLookupPredicates(offeredPredicates.clone())) + residual ++= unselectedPredicates.map(_.condition) + + val scan = scanBuilder.build() + val resolver = SQLConf.get.resolver + // Fields match the way ordinary V2 scans match them (`PushDownUtils.toOutputAttrs`): CHAR and + // VARCHAR are read as STRING, and nullability, including nested nullability, does not count. + val readSchema = CharVarcharUtils.replaceCharVarcharWithStringInSchema(scan.readSchema()) + val lookupInput = readSchema.fields.map { field => + relation.output.find { attr => + resolver(attr.name, field.name) && + DataType.equalsIgnoreNullability(attr.dataType, field.dataType) + }.getOrElse { + throw QueryCompilationErrors.invalidDataSourceExternalLookupError( + tableName, "UNKNOWN_SCAN_FIELD", Map("field" -> toSQLId(Seq(field.name)))) + } + }.toSeq + lookupInput.groupBy(_.name).collectFirst { + case (name, fields) if fields.size > 1 => name + }.foreach { name => + throw QueryCompilationErrors.invalidDataSourceExternalLookupError( + tableName, "DUPLICATE_SCAN_FIELD", Map("field" -> toSQLId(Seq(name)))) + } + val externalLookup = scan.toExternalLookup() + val maxParameterRowsPerRequest = externalLookup.maxParameterRowsPerRequest() + if (maxParameterRowsPerRequest <= 0) { + throw QueryCompilationErrors.invalidDataSourceExternalLookupError( + tableName, + "NON_POSITIVE_MAX_PARAMETER_ROWS", + Map("value" -> maxParameterRowsPerRequest.toString)) + } + val readerFactory = externalLookup.createReaderFactory() + // TODO: prune lookup output to the columns required above the join. + val lookupOutput = lookupInput Review Comment: **Non-blocking (P2):** `lookupOutput` is `lookupInput`, which has one attribute per `scan.readSchema()` field in read-schema order. Nothing compares it with `requiredLookupOutput` or the analyzed `right.output`, and nothing restores the analyzed order, as `V2ScanRelationPushDown` does with its re-projection for ordinary scans. `Scan.toExternalLookup` only requires top-level table columns, each at most once, so a compliant connector may return `(value, id)` or just `(value)` for a table `(id, value)`. With a reordered read schema, `PreparedExternalLookupJoin.output` differs from the `ExternalLookupJoin` output it replaces. At the plan root the result schema changes between analysis and optimization: plan-change validation catches this in tests, while consumers that bind by position, such as Dataset deserializers and sinks, read the wrong columns. With a partial read schema, parents keep references to the dropped columns. Because `PreparedExternalLookupJoin.references` intersects the residual's references with `child.outputSet`, `missingInput` does not expose a residual reference to an omitted lookup column either, so the failure surfaces only at binding, as an internal error rather than a `DATA_SOURCE_INVALID_EXTERNAL_LOOKUP` error. The suite's `lookup.output === join.output` holds only because its stub returns the table schema in table order. This is reachable once the rule is registered. See **Shared repair plan 1** in the review body. ########## common/utils/src/main/resources/error/error-conditions.json: ########## @@ -2817,6 +2861,59 @@ ], "sqlState" : "42822" }, + "EXTERNAL_LOOKUP_JOIN_UNSUPPORTED" : { + "message" : [ + "The join cannot be executed as an external lookup join:" + ], + "subClass" : { + "CONFLICTING_JOIN_STRATEGY" : { + "message" : [ + "The join strategy hint <strategy> cannot be honored by an external lookup join over <relation>. Remove the <strategy> hint, or the `EXTERNAL_LOOKUP` hint if the table also supports batch reads." + ] + }, + "JOIN_TYPE" : { + "message" : [ + "The join type <joinType> is not supported by an external lookup join over <relation>. Supported join types are <supportedJoinTypes>." + ] + }, + "LOOKUP_PLAN" : { + "message" : [ + "The lookup side must be a DSv2 table, optionally followed by projections and filters." + ] + }, + "LOOKUP_TABLE_ON_LEFT" : { + "message" : [ + "The lookup-only table <relation> must be on the right side of the join." + ] + }, + "MISSING_JOIN_CONDITION" : { + "message" : [ + "The join has no join condition." + ] + }, + "NO_LOOKUP_PREDICATE" : { + "message" : [ + "The data source of table <table> did not select any lookup predicate from the join condition." Review Comment: **Nit (P3):** When no conjunct has the shape `lookupColumn = <stream-side expression>`, `selectLookupPredicates` receives an empty array, a compliant connector returns an empty array, and this message tells the user that the data source did not select any lookup predicate. That is reachable with an ordinary equality: for a BIGINT stream key and an INT lookup column, `ON t.id = s.id` is coerced to `s.id = CAST(t.id AS BIGINT)`, and neither `EqualTo` arm in `prepare` accepts a cast on the lookup side. The same applies to range or `<=>` conditions, which this layer defers, and to expressions over a lookup column. Nothing in the error says that Spark offers only equalities on a bare lookup column, or that casting the stream side instead would make the equality offerable. Naming that shape in the message, or failing before the connector call with a separate reason when nothing was offered, would point users at the actual cause. See **Shared repair plan 2** in the review body. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PrepareExternalLookupJoin.scala: ########## @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.datasources.v2 + +import java.util.{Collections, IdentityHashMap} + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualTo, Expression, + PredicateHelper} +import org.apache.spark.sql.catalyst.plans.logical.{ExternalLookupJoin, Filter, Join, LogicalPlan, + PreparedExternalLookupJoin, Project} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.EXTERNAL_LOOKUP_JOIN +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, FieldReference} +import org.apache.spark.sql.connector.expressions.filter.{Predicate => V2Predicate} +import org.apache.spark.sql.connector.read.{LookupParameter, SupportsExternalLookup} +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.DataType + +/** Finalizes an optimized external lookup side immediately before ordinary V2 scan preparation. */ +object PrepareExternalLookupJoin extends Rule[LogicalPlan] with PredicateHelper + with QueryErrorsBase { + override def apply(plan: LogicalPlan): LogicalPlan = { + val prepared = plan.transformUpWithPruning(_.containsPattern(EXTERNAL_LOOKUP_JOIN)) { + case join: ExternalLookupJoin => prepare(join) + } + // Prepared lookup joins no longer contain their lookup relation, so a lookup-only table that + // is still read here cannot run. Pre-order visits a join before the relation it reads. + prepared.foreach { + // Classic Dataset.join(right, usingColumns) first analyzes Join(left, right, joinType, None), + // including CheckAnalysis, to disambiguate a self-join, and only then builds the USING join. + // That plan cannot be told apart from a join without a condition, so analysis keeps it, and + // a missing condition is reported here, on the final plan. + case Join(_, right, _, None, _) if readsLookupOnlyTable(right) => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("MISSING_JOIN_CONDITION") + case node => + ExternalLookupRelations.lookupOnlyTable(node).foreach { table => + throw QueryCompilationErrors.unsupportedBatchReadError(table) + } + } + prepared + } + + private def readsLookupOnlyTable(plan: LogicalPlan): Boolean = plan match { + case node @ (_: Project | _: Filter) => readsLookupOnlyTable(node.children.head) + case node => ExternalLookupRelations.lookupOnlyTable(node).isDefined + } + + private def prepare(join: ExternalLookupJoin): PreparedExternalLookupJoin = { + val relation = extractLookupPlan(join.right).getOrElse { + throw QueryCompilationErrors.unsupportedExternalLookupJoinError("LOOKUP_PLAN") + } + val tableName = relation.table.name() + val supportsRead = relation.table match { + case table: SupportsRead + if table.capabilities().contains(TableCapability.EXTERNAL_LOOKUP) => table + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "TABLE", Map("table" -> toSQLId(tableName))) + } + val scanBuilder = supportsRead.newScanBuilder(relation.options) + val lookupBuilder = scanBuilder match { + case builder: SupportsExternalLookup => builder + case _ => + throw QueryCompilationErrors.unsupportedExternalLookupJoinError( + "SCAN_BUILDER", Map("table" -> toSQLId(tableName))) + } + + val conjuncts = join.condition.toSeq.flatMap(splitConjunctivePredicates) + val offeredCandidates = mutable.ArrayBuffer.empty[OfferedLookupPredicate] + val residual = mutable.ArrayBuffer.empty[Expression] + conjuncts.foreach { + case equality @ EqualTo(leftExpr, rightAttr: Attribute) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case equality @ EqualTo(rightAttr: Attribute, leftExpr) + if leftExpr.references.subsetOf(join.left.outputSet) && + join.right.outputSet.contains(rightAttr) => + offeredCandidates += OfferedLookupPredicate(leftExpr, rightAttr, equality, "=", false) + case other => residual += other + } + + val offeredPredicates = offeredCandidates.map { candidate => + new V2Predicate(candidate.predicateName, Array[V2Expression]( + FieldReference.column(candidate.projectedField.name), + new LookupParameter(candidate.parameter.dataType))) + }.toArray + // A copy, so a connector that reorders the array in place cannot break the identity mapping. + val (selectedPredicates, unselectedPredicates) = selectedCandidates(tableName, + offeredPredicates, offeredCandidates, + lookupBuilder.selectLookupPredicates(offeredPredicates.clone())) + residual ++= unselectedPredicates.map(_.condition) + + val scan = scanBuilder.build() + val resolver = SQLConf.get.resolver + // Fields match the way ordinary V2 scans match them (`PushDownUtils.toOutputAttrs`): CHAR and + // VARCHAR are read as STRING, and nullability, including nested nullability, does not count. + val readSchema = CharVarcharUtils.replaceCharVarcharWithStringInSchema(scan.readSchema()) + val lookupInput = readSchema.fields.map { field => + relation.output.find { attr => Review Comment: **Nit (P3):** The matcher ignores nullability (`DataType.equalsIgnoreNullability`), so a read schema that declares `value STRING` for a `value STRING NOT NULL` column, or `s STRUCT<a INT>` for `s STRUCT<a INT NOT NULL>` (the PR's own example), is accepted. `lookupInput` still reuses the relation attribute, with the table's nullability, although `ExternalLookupResultCursor.get()` returns rows whose schema is `Scan#readSchema()`. Once an execution layer reads value rows through `lookupInput`, a null that the read schema permits is read without a null check: a null INT reads as 0 and a null string can cause an NPE inside the lookup operator. Keeping the analyzed nullability in the prepared output is consistent with ordinary V2 scans: operator optimization has already relied on it before this rule runs, and the ordinary restoring Project keeps the analyzed top-level nullability too. The gap is the row description itself. Under an inner join the plan above trusts the table's NOT NULL anyway, but under a left outer join every operator above treats the lookup columns as nullable, and `lookupInput` is the only place that does not. Building `lookupInput` from the read schema the way `toOutputAttrs` builds scan attributes, with the scan's type and nullability and the relation's exprId and CHAR/VARCHAR metadata, while `lookupOutput` keeps the analyzed attributes, closes it. Rejecting read schemas with weaker nullability is the alternative. See **Shared repair plan 1** in the review body. ########## sql/core/src/test/scala/org/apache/spark/sql/connector/PrepareExternalLookupJoinSuite.scala: ########## @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector + +import org.apache.spark.SparkException +import org.apache.spark.sql.AnalysisException +import org.apache.spark.sql.QueryTest +import org.apache.spark.sql.catalyst.analysis.ResolveExternalLookupJoin +import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, EqualTo, Literal} +import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftOuter} +import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, LocalRelation, + PreparedExternalLookupJoin, Project} +import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability} +import org.apache.spark.sql.connector.catalog.TableCapability.EXTERNAL_LOOKUP +import org.apache.spark.sql.connector.expressions.FieldReference +import org.apache.spark.sql.connector.expressions.filter.Predicate +import org.apache.spark.sql.connector.read.{ExternalLookup, ExternalLookupReaderFactory, Scan, + ScanBuilder, SupportsExternalLookup} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, + PrepareExternalLookupJoin} +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.{IntegerType, StructType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +class PrepareExternalLookupJoinSuite extends QueryTest with SharedSparkSession { + + test("a lookup-only relation that no lookup join reads is rejected") { + val left = LocalRelation(Seq(AttributeReference("id", IntegerType)()), isStreaming = true) + val relation = DataSourceV2Relation.create( + PlanningExternalLookupTable, None, None, CaseInsensitiveStringMap.empty) + val lookupJoin = ResolveExternalLookupJoin(Join( + left, relation, Inner, Some(EqualTo(left.output.head, relation.output.head)), JoinHint.NONE)) + + assert(PrepareExternalLookupJoin(lookupJoin).isInstanceOf[PreparedExternalLookupJoin]) + checkError( + exception = intercept[AnalysisException] { + PrepareExternalLookupJoin(Project(relation.output, relation)) + }, + condition = "UNSUPPORTED_FEATURE.TABLE_OPERATION", + parameters = Map( + "tableName" -> "`PlanningExternalLookupTable`", + "operation" -> "batch scan")) + } + + test("PrepareExternalLookupJoin binds the selected predicates and keeps the rest as residual") { + val left = LocalRelation(Seq(AttributeReference("id", IntegerType)()), isStreaming = true) + val relation = DataSourceV2Relation.create( + PlanningExternalLookupTable, None, None, CaseInsensitiveStringMap.empty) + val Seq(id, value) = relation.output + // Offered to the connector as `value = ?`, but not selected. + val unselected = EqualTo(value, Literal("a")) + val join = Join(left, relation, LeftOuter, Some(And(EqualTo(left.output.head, id), unselected)), + JoinHint.NONE) + + val prepared = PrepareExternalLookupJoin(ResolveExternalLookupJoin(join)) + prepared match { + case lookup: PreparedExternalLookupJoin => + assert(lookup.lookupParameters === left.output) + assert(lookup.residualCondition === Some(unselected)) + assert(lookup.output === join.output) + checkError( + exception = intercept[SparkException] { lookup.copy(joinType = FullOuter).output }, + condition = "INTERNAL_ERROR", + parameters = Map("message" -> "Unexpected external lookup join type FullOuter")) + case other => fail(s"Expected a prepared external lookup join: $other") + } + assert(PrepareExternalLookupJoin(prepared) === prepared) + } + + test("a join without a condition over a lookup-only table fails on the final plan") { + val left = LocalRelation(Seq(AttributeReference("id", IntegerType)()), isStreaming = true) + val relation = DataSourceV2Relation.create( + PlanningExternalLookupTable, None, None, CaseInsensitiveStringMap.empty) + // Analysis keeps such a join, because the Dataset API analyzes a USING join first as one. + checkError( + exception = intercept[AnalysisException] { + PrepareExternalLookupJoin(ResolveExternalLookupJoin( + Join(left, relation, Inner, None, JoinHint.NONE))) + }, + condition = "EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.MISSING_JOIN_CONDITION", + parameters = Map.empty) + } +} + +/** A lookup-only table that the planning rules can prepare. It is never read. */ +object PlanningExternalLookupTable extends Table with SupportsRead { Review Comment: **Nit (P3):** `PlanningExternalLookupTable` is the only connector in the suite, and it never misbehaves: it selects every `id = ?` candidate, returns `schema()` as `readSchema()`, and reports a request size of 1. So `TABLE`, `SCAN_BUILDER`, `LOOKUP_PLAN`, `NO_LOOKUP_PREDICATE` and every `DATA_SOURCE_INVALID_EXTERNAL_LOOKUP` subclass are unreached. So are CHAR/VARCHAR normalization, ignored nullability, resolver case handling, binding when two or more candidates are selected in a different order than offered, which conjuncts are offered and in which form, and the redacted `stringArgs`. Binding parameters in offered order, dropping the defensive clone, removing the null-selection check (a null element would then reach `predicate.describe()` and throw an NPE), or widening `stringArgs` to include the scan would all leave the suite green. Configurable stubs like this one can cover them at plan level today; only end-to-end execution needs SPARK-60690. **Verification:** - **Regression:** Binding parameters in offered order instead of selected order, or removing the null-selection check, makes a test fail. - **Behavior:** For each DATA_SOURCE_INVALID_EXTERNAL_LOOKUP subclass and each preparation-time EXTERNAL_LOOKUP_JOIN_UNSUPPORTED subclass, a stub connector that violates the matching contract raises it with its documented parameters. - **Regression:** Rendering a prepared lookup join whose scan has a distinctive description does not include that description. -- 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]
