HeartSaVioR commented on code in PR #59909:
URL: https://github.com/apache/spark/pull/59909#discussion_r4419731803
##########
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:
> [!NOTE]
> **DISCLAIMER:** This comment was written by an LLM, Claude Opus 5.5
(`claude-opus-5-5`, 1M context), posting on behalf of @HeartSaVioR.
Fixed in
[73b62b7c559](https://github.com/apache/spark/commit/73b62b7c559b90caf5ac9088417679591118c5d5).
The message now reads "The lookup scan returned the field <field>, which
matches no table column by name and data type." The suite tests the
type-mismatch case (`value INT` for a `STRING` column).
##########
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:
> [!NOTE]
> **DISCLAIMER:** This comment was written by an LLM, Claude Opus 5.5
(`claude-opus-5-5`, 1M context), posting on behalf of @HeartSaVioR.
Fixed in
[73b62b7c559](https://github.com/apache/spark/commit/73b62b7c559b90caf5ac9088417679591118c5d5),
using the second option. When no conjunct has the form `lookupColumn =
<stream-side expression>` with a bare lookup column, the rule fails with the
new `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.NO_LOOKUP_CANDIDATE` before it calls the
connector, including before `newScanBuilder`. The message names the accepted
form. It also says that when the lookup column is cast to match the stream
side, the stream-side expression should be cast to the lookup column's type
instead. `NO_LOOKUP_PREDICATE` is now raised only when the connector selects
none of a non-empty offer, and its text says the predicates were offered. The
test joins a BIGINT stream key with an INT lookup column (`s.id = CAST(t.id AS
BIGINT)`) through a stub that fails the test if its selection is called.
##########
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:
> [!NOTE]
> **DISCLAIMER:** This comment was written by an LLM, Claude Opus 5.5
(`claude-opus-5-5`, 1M context), posting on behalf of @HeartSaVioR.
Fixed in
[73b62b7c559](https://github.com/apache/spark/commit/73b62b7c559b90caf5ac9088417679591118c5d5).
The suite now extends `PlanTest` and doesn't start a session.
##########
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:
> [!NOTE]
> **DISCLAIMER:** This comment was written by an LLM, Claude Opus 5.5
(`claude-opus-5-5`, 1M context), posting on behalf of @HeartSaVioR.
Partly addressed in
[73b62b7c559](https://github.com/apache/spark/commit/73b62b7c559b90caf5ac9088417679591118c5d5).
The table stub now takes a selection function and a read schema. The suite
uses it to cover several candidates selected in a different order than offered,
an unselected candidate kept in the residual in written order, a reordered read
schema with weaker top-level and nested nullability, columns that differ only
by case, a missing column, a type mismatch, a condition with no offerable
candidate, an empty selection, and a null selection or a null element. Binding
parameters in offered order, removing the null-selection check, or reordering
the residual now fails a test. The other contract errors are already raised end
to end with their parameters in SPARK-60690's `ExternalLookupJoinSuite`
(`UNOFFERED_SELECTED_PREDICATE`, `DUPLICATE_SELECTED_PREDICATE`,
`NON_POSITIVE_MAX_PARAMETER_ROWS`, `UNKNOWN_SCAN_FIELD`,
`DUPLICATE_SCAN_FIELD`, `TABLE`, `SCAN_BUILDER`, `NO_LOOKUP_P
REDICATE`), so they aren't repeated here. The scan description is redacted by
the physical node, which is added and tested in SPARK-60690.
##########
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:
> [!NOTE]
> **DISCLAIMER:** This comment was written by an LLM, Claude Opus 5.5
(`claude-opus-5-5`, 1M context), posting on behalf of @HeartSaVioR.
Partly addressed in
[73b62b7c559](https://github.com/apache/spark/commit/73b62b7c559b90caf5ac9088417679591118c5d5).
Both suites now use `ExternalLookupRelation` as the lookup side. In
`PrepareExternalLookupJoinSuite`, that covers its classification, its
unwrapping in `extractLookupPlan`, and the rejection of a stray
`ExternalLookupRelation` outside a lookup join. A new test runs `ColumnPruning`
over a projection of stream columns above the lookup join. It checks that the
stream side is pruned, the lookup relation stays directly under the join, and
the join still prepares. Without `requiredLookupOutput` in the references, that
test fails. The hint tests now run the analyzer, so a strategy hint is a
`ResolvedHint` node above the stream side, as it is when the conversion runs.
Both the analysis-time check and the conversion raise
`CONFLICTING_JOIN_STRATEGY` for it. A new test puts a `broadcast` hint inside a
nested join on the stream side and checks that the lookup join still forms,
because the hint belongs to the nested join.
--
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]
