[SPARK-59342][SQL] Allow configuring schema alignment behavior for DSv2 writes

[johanl-db]

What changes were proposed in this pull request?

This change extends the DSv2 connector interface to allow connectors to configure schema alignment behavior. For DSv2 batch / row-level writes, under spark.sql.storeAssignmentPolicy=ANSI, implicit cast are validated at analysis time, rejecting casts that are valid but may fail, e.g. STRING->INT.

This change allows connectors to opt into runtime checks instead of analysis check: allow more casts during analysis (STRING-> any, complex type -> STRING, ..) and fail at runtime on malformed / overflowing values.

New internal DSV2 interface:

public interface ConfigurableSchemaAlignment extends Table {
  SchemaAlignmentConfig schemaAlignmentConfig();
}

public interface SchemaAlignmentConfig {
  default AnsiStoreAssignmentCastCheck ansiStoreAssignmentCastCheck() {
    // AT_ANALYSIS or AT_RUNTIME.
    return AnsiStoreAssignmentCastCheck.AT_ANALYSIS;
  }
}

Why are the changes needed?

This allows connector to configure and opt out of strict schema alignement behavior when needed.

How was this patch tested?

Add Spark suite SchemaAlignmentConfigSuite covering the configuration mechanism

[gengliangwang]

Requesting changes for two correctness blockers: deferred ANSI mode bypasses structural validation and can silently corrupt UDT-backed structs, and the LEGACY opt-in is not enforced for already-aligned writes. The public contract, versioning, state propagation, and tests also need tightening. The unrelated InsertSchemaEvolutionSuite should be split from this PR, and its broad intercept[Exception] should assert the expected error instead.

[gengliangwang]

Review summary

The latest revision fixes the earlier structural-validation bypass, configuration propagation, API version, and permissive test helper. One blocking analyzer issue remains: opted-in LEGACY writes with nullable expressions never satisfy the existing completion predicates and exhaust analyzer iterations. I also found three additional correctness or test gaps and three public-contract defects; two reuse the existing discussions.

Findings

7 total: 0 P0, 1 P1, 4 P2, 2 P3.

Blocking (P1)

• Make LEGACY nullability compatible with analyzer completion — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:3991 — see inline.

Non-blocking (P2)

• Narrow runtime-deferral documentation to atomic ANSI checks — docs/sql-ref-ansi-compliance.md:230 — see inline.
• Snapshot row-level alignment options once per command — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala:145 — see inline.
• Test malformed and overflowing deferred ANSI values — sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala:125 — already raised in an existing discussion.
• Apply LEGACY policy to static partition casts — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:3991
  The table opt-in makes LEGACY DSv2 inserts supported, but static partition values are still created with ansiEnabled = true and static-overwrite predicates use the session ANSI cast mode. A malformed or overflowing partition literal therefore fails instead of following LEGACY's null-partition behavior, and changing only the projection could make overwrite select a different partition. Please bind both casts to the validated table policy.

Nit (P3)

• Use the verb phrase opt out of — sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java:38 — see inline.
• Scope the Table hook to batch and row-level writes — sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java:117 — already raised in an existing discussion.

Existing discussions

• Suppressed duplicate: Test malformed and overflowing deferred ANSI values — P2 at sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala:125 — existing discussion
• Suppressed duplicate: Scope the Table hook to batch and row-level writes — P3 at sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java:117 — existing discussion
• existing discussion — The earlier review's UDT structural bypass, config-copy loss, streaming scope, @SInCE value, and weak outcome helper were repaired. Its exact aligned-write gate concern remains in the head but is preserved merge-base behavior and therefore is not bound as a finding. The current nullable-LEGACY fixed-point failure and overbroad public documentation are related new defects on the review's stated public-contract and test surfaces, with different mechanisms and correction boundaries.

PR description suggestions

• Refresh the PR body's API snippet to use the current deferAnsiCastValidationToRuntime method name and describe the hook as applying to DSv2 batch and row-level writes, not streaming.

[szehon-ho]

I was actually wondering if there is a plan to go through in future myself

[johanl-db]

Not that I know of.
The interplay between schema alignment and schema evolution is particularly interesting, I suspect if we ever want to support schema evolution in DSv2 streaming, we may have to implement schema alignment also.

[johanl-db]

@gengliangwang @szehon-ho thanks for the reviews

I applied the following substantial changes:

• Completely removed allowing LEGACY: this isn't well supported in DSv2 and would require additional work not fitting in this PR. Leaving as a following if it can be reasonably supported
• Re-structured the interface: now an internal ConfigurableSchemaAlignment interface that can be added to table instead of a default method on Table itself.
• Finer-grain cast check to properly validate e.p. struct field ordering and UDTs.