gengliangwang commented on code in PR #58632:
URL: https://github.com/apache/spark/pull/58632#discussion_r4023002484
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3981,9 +3985,10 @@ class Analyzer(
}
}
- private def validateStoreAssignmentPolicy(): Unit = {
- // SPARK-28730: LEGACY store assignment policy is disallowed in data
source v2.
- if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY) {
+ private def validateStoreAssignmentPolicy(schemaAlignment:
SchemaAlignmentConfig): Unit = {
+ // SPARK-28730: LEGACY store assignment policy is disallowed in data
source v2 by default.
+ if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY &&
+ !schemaAlignment.allowLegacyStoreAssignmentPolicy()) {
Review Comment:
**Blocking (P1):** Once this gate admits an opted-in LEGACY table, nullable
expressions can reach alignment. LEGACY intentionally omits AssertNotNull, but
V2WriteCommand.outputResolved and AssignmentUtils.aligned still require a
non-nullable result, so INSERT, UPDATE, and MERGE keep rebuilding aliases until
analyzer iteration exhaustion.
**Recommended change:** Make batch output and row-level assignment
completion policy-aware for a table that has passed the LEGACY opt-in gate,
while preserving truthful nullable expression metadata and all ANSI and STRICT
null checks.
**Why this works:** Carry the validated command-level alignment policy to
the compatibility and aligned checks so LEGACY accepts the nullable result it
deliberately constructs; do not use AssertNotNull or falsify expression
nullability merely to satisfy the old predicate.
**Scope:** Reconcile LEGACY expression nullability with DSv2 write-command
fixed-point completion across INSERT, UPDATE, and MERGE.
**Compatibility:** LEGACY omits the ANSI and STRICT runtime null assertion,
while the other store-assignment policies remain unchanged.
**Risks:** A global nullability relaxation could weaken ANSI or STRICT
writes. Checking completion before validating the table opt-in could preserve
or widen the pre-existing aligned-write policy bypass.
**Constraints:** DEFAULT tables must retain the existing LEGACY rejection on
paths that perform alignment. ANSI and STRICT must retain AssertNotNull and
compatible-nullability completion semantics. Nullable expression metadata must
remain truthful.
**Success:** An opted-in LEGACY INSERT from a nullable same-type or cast
expression into a NOT NULL target reaches a fixed point. Opted-in LEGACY UPDATE
and every assignment-bearing MERGE arm reach aligned rewritable plans for
nullable inputs. ANSI, STRICT, and non-opted-in behavior retain their existing
nullability checks and errors.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala:
##########
@@ -137,7 +141,8 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
val value = matchingAssignments.head.value
val coerceMode = if (coerceNestedTypes) RECURSE else NONE
TableOutputResolver.resolveUpdate(
- "", value, actualAttr, conf, err => errors += err, colPath,
coerceMode)
+ "", value, actualAttr, conf, err => errors += err, colPath,
coerceMode,
+ deferAnsiCastValidationToRuntime =
schemaAlignment.deferAnsiCastValidationToRuntime())
Review Comment:
**Non-blocking (P2):** Row-level alignment calls this getter separately for
fields, nested leaves, and MERGE actions. SchemaAlignmentConfig does not
require immutable getters, so a connector-backed live flag can make one command
apply mixed policy and become interleaving-dependent.
**Recommended change:** Evaluate the schema-alignment Boolean options once
for each row-level command and pass immutable values through every assignment,
nested field, and MERGE action alignment call.
**Why this works:** Replace recursive reads of SchemaAlignmentConfig with
primitive command-level policy values or an internal immutable snapshot
produced when the table config is obtained; retain the public SPI unchanged.
**Scope:** Give one row-level command a coherent schema-alignment policy
across all fields and actions.
**Compatibility:** Connector-specific option values still control each newly
analyzed command; only intra-command rereads are removed.
**Risks:** Snapshotting at each action rather than at command setup would
still permit mixed UPDATE and INSERT behavior within one MERGE.
**Constraints:** All assignment-bearing actions in one MERGE must share the
same snapshot. The public SchemaAlignmentConfig API and existing immutable
implementations must remain source-compatible.
**Success:** Every field and action in one row-level command observes one
deferral value. A later connector-side flag change affects a later analysis
command, not part of the current command.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.catalog;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Schema-alignment configuration for batch/row-level writes to a {@link
Table}. This allows
+ * connectors to configure casting behavior and handling of schema mismatches
during DSv2 writes.
+ * It is not consulted for streaming writes, which do not go through this
alignment path.
+ *
+ * @since 4.4.0
+ */
+@Evolving
+public interface SchemaAlignmentConfig {
+
+ /** The strict data source v2 configuration, returned by {@link Table} by
default. */
+ SchemaAlignmentConfig DEFAULT = new SchemaAlignmentConfig() {};
+
+ /**
+ * Whether {@code spark.sql.storeAssignmentPolicy=LEGACY} is allowed for
writes and row-level
+ * operations targeting this table. Data source v2 rejects LEGACY by
default; a table can decide
+ * to opt-out from this restriction.
Review Comment:
**Nit (P3):** This public Javadoc should say `opt out of this restriction`;
`opt-out` is the noun or adjective form, and the verb takes `of`.
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -226,6 +226,9 @@ INSERT INTO test VALUES (2147483648L);
org.apache.spark.SparkArithmeticException: [CAST_OVERFLOW_IN_TABLE_INSERT]
Fail to insert a value of "BIGINT" type into the "INT" type column `i` due to
an overflow. Use `try_cast` on the input value to tolerate overflow and return
NULL instead.
```
+By default, invalid source/target combinations are rejected during analysis.
+Data sources may instead defer this validation to execution time, so an
insertion is rejected only when a value is actually malformed or overflows, not
during analysis.
Review Comment:
**Non-blocking (P2):** This wording makes the deferral sound universal, but
the implementation only relaxes the atomic ANSI store-assignment check.
Structural mismatches and pairs Spark cannot cast still fail during analysis,
so please state that boundary explicitly.
--
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]
