HeartSaVioR commented on code in PR #59909: URL: https://github.com/apache/spark/pull/59909#discussion_r4419732544
########## 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/pull/59909/commits/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_PREDICATE`), 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/pull/59909/commits/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 s till 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]
