[SPARK-59275][PYTHON] Complete CHAR/VARCHAR support for Python UDFs and Arrow

[srielau]

What changes were proposed in this pull request?

Complete first-class CHAR/VARCHAR support at PySpark and Arrow boundaries when
spark.sql.charVarchar.standardSemantics.enabled is true:

• Map Python CharType and VarcharType to Arrow UTF8, recursively through complex types.
• Apply CHAR padding and VARCHAR length checks to pickled, Arrow-optimized, and pandas UDF
  results before they re-enter Catalyst.
• Compare Arrow UDF output using its physical STRING representation while retaining logical
  CHAR/VARCHAR in the Spark schema.
• Apply the same recursive assignment checks to local, pandas, and PyArrow explicit-schema
  DataFrame creation.
• Allow DataFrame.toArrow() to export CHAR/VARCHAR values as Arrow strings.
• Handle CHAR/VARCHAR output in the Arrow columnar-input execution path.

JIRA: https://issues.apache.org/jira/browse/SPARK-59275

Why are the changes needed?

CHAR/VARCHAR are first-class types under standard semantics, but Python and Arrow boundaries
still treated them inconsistently. Arrow UDFs rejected them as unsupported, pickled UDF and
explicit-schema creation paths did not enforce their length rules, and Arrow output validation
compared logical CHAR/VARCHAR against physical STRING.

These gaps allowed unpadded CHAR and over-length VARCHAR values or caused supported queries to
fail. The checks must also recurse through structs, arrays, and maps.

Does this PR introduce any user-facing change?

Yes. With spark.sql.charVarchar.standardSemantics.enabled=true, Python UDFs, Arrow-optimized
UDFs, pandas UDFs, and explicit-schema DataFrame creation now accept CHAR/VARCHAR and enforce
their assignment semantics. toArrow() exports these values as Arrow strings. The default
flag-off behavior is unchanged.

How was this patch tested?

Added PySpark coverage for:

• Non-Arrow, Arrow-optimized, and pandas UDF CHAR/VARCHAR results.
• CHAR padding, VARCHAR overflow, and nested struct/array/map results.
• Arrow columnar input.
• Local, pandas, and PyArrow explicit-schema DataFrame creation.
• DataFrame.toArrow().

Ran:

/usr/bin/sbt -java-home /usr/lib/jvm/java-17-openjdk-amd64 \
  -Dsbt.override.build.repos=true 'sql/Test/compile'

/usr/bin/sbt -java-home /usr/lib/jvm/java-17-openjdk-amd64 \
  -Dsbt.override.build.repos=true \
  'sql/testOnly org.apache.spark.sql.execution.python.EvaluatePythonSuite -- -z SPARK-59275'

python3 -m py_compile \
  python/pyspark/sql/pandas/types.py \
  python/pyspark/sql/tests/test_udf.py \
  python/pyspark/sql/tests/test_creation.py \
  python/pyspark/sql/tests/arrow/test_arrow.py \
  python/pyspark/sql/tests/arrow/test_arrow_python_udf.py

The PySpark integration tests were not run locally because PyArrow is not installed in this
environment.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Cursor Auto (GPT-5.6)

[cloud-fan]

Review summary

The current implementation is not ready because it widens non-scalar pandas/Arrow UDF APIs without adding their required JVM assignment checks, and it changes the documented legacy CHAR/VARCHAR-as-STRING behavior. These can expose invalid logical values or introduce new runtime failures in supported configurations. The large-input Arrow RDD path also gains an unconditional per-row projection/copy, and the added Parquet test does not exercise the new Arrow-backed columnar CHAR/VARCHAR branch.

Findings

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

Blocking (P1)

• Honor legacy CHAR/VARCHAR-as-STRING mode — sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:212 — see inline.
• Do not enable unchecked non-scalar Arrow outputs — python/pyspark/sql/pandas/types.py:138 — see inline.

Non-blocking (P2)

• Exercise the Arrow-backed CHAR/VARCHAR branch — python/pyspark/sql/tests/arrow/test_arrow_python_udf.py:316 — see inline.
• Keep the direct Arrow RDD path for unconstrained schemas — sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:570 — see inline.

Verification

• The changed shared Arrow mapper is called by public non-scalar UDF return-type validation, while their unchanged JVM consumers do not apply stringLengthCheck.
• The added Parquet case and the changed ArrowColumnVector guard select different evaluator paths.

PR metadata suggestions

• Narrow the blanket pandas UDF support claim to scalar UDFs unless assignment checks and tests are added for map, grouped, cogrouped, and aggregate eval types.
• Do not claim Arrow columnar-input coverage until a test uses ArrowBackedDataSourceV2 and exercises the ArrowColumnVector CHAR/VARCHAR output branch.

[cloud-fan]

Review summary

The change is not ready because the capability boundary is still inconsistent: two public grouped-aggregate iterator eval types bypass rejection, Arrow UDTFs are newly admitted without a compatible JVM consumer, and fused nested UDFs skip the declared intermediate CHAR/VARCHAR boundary. These paths can expose unpadded or over-length values, fail valid Arrow UDTF output, or make nested UDF results depend on fusion. The row and Arrow-columnar legacy branches also need focused tests for unchanged legacy values and, for the columnar case, path selection.

Findings

5 total: 0 P0, 3 P1, 2 P2, 0 P3.

Blocking (P1)

• Reject CHAR/VARCHAR for iterator grouped aggregates — python/pyspark/sql/udf.py:329 — see inline.
• Preserve CHAR/VARCHAR semantics across fused UDFs — sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvalPythonEvaluatorFactory.scala:44 — see inline.
• Do not partially activate CHAR/VARCHAR for Arrow UDTFs — python/pyspark/sql/pandas/types.py:138 — see inline.

Non-blocking (P2)

• Cover legacy semantics on Arrow scalar UDF output — python/pyspark/sql/tests/arrow/test_arrow_python_udf.py:286 — see inline.
• Cover legacy semantics on the Arrow-columnar path — sql/core/src/test/scala/org/apache/spark/sql/execution/python/ArrowColumnarPythonUDFSuite.scala:112 — see inline.

Re-review status

Prior AI findings: 3 addressed, 1 still present; additional unresolved findings in this review: 4.

New attribution: 2 newly introduced, 2 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

• Reject CHAR/VARCHAR for iterator grouped aggregates — python/pyspark/sql/udf.py:329

Existing discussions

• existing discussion — The current head rejects the originally listed non-iterator map, grouped-map, cogrouped-map, and aggregate variants, but leaves the two public grouped-aggregate iterator variants outside the rejection dispatch.

Verification

• Both omitted grouped-aggregate iterator eval types are public and supported by ArrowAggregatePythonExec, whose output projection does not call stringLengthCheck.
• Both scalar evaluators fuse nested same-eval-type UDFs and apply constrained-string checks only to final PythonExec output attributes.
• The shared Python mapping emits Arrow strings for CHAR/VARCHAR, while ArrowEvalPythonUDTFExec compares logical result attributes directly with physical Arrow column types and has no write-side projection.

PR metadata suggestions

• Narrow the title and the broad claims of complete Python UDF and Arrow support to the scalar UDF, DataFrame creation, and toArrow surfaces that have matching consumers; explicitly state that non-scalar eval types and Arrow UDTFs remain unsupported until their assignment semantics are implemented.
• Update the testing section to disclose that fused intermediate UDF results and the row and Arrow-columnar legacy-as-string branches are not covered; for the columnar branch, call out the missing full-columnar path-selection assertion.

[zhengruifeng]

Could you clarify whether a new configuration is necessary, given that CHAR/VARCHAR previously failed at these Arrow boundaries and existing configurations already define their semantics?
Could you also audit all to_arrow_type/to_arrow_schema call sites, particularly Arrow UDTF and Python DataSource, to ensure they either remain unsupported or handle physical STRING normalization and assignment checks?

[cloud-fan]

Review summary

The patch is not ready in its pinned form. The recursive type helpers and most boundary-specific tests are sound, but five supported execution paths still violate the capability or configuration owner that determines CHAR/VARCHAR semantics: unhandled incremental/stateful eval types and analyze-derived Arrow UDTFs bypass rejection, a constrained-string sibling crashes supported UDT output in the columnar fallback, and state-server plus persisted-view execution consult ambient SQLConf instead of query-bound semantics. The remaining P2 issues are an unnecessary repeated Connect parse, leaked local Arrow resources after validation failure, duplicate nested result traversal, and missing execution coverage for three admitted scalar protocols. This was a static review; I did not run tests.

Findings

9 total: 0 P0, 5 P1, 4 P2, 0 P3.

Blocking (P1)

• Make Python eval-type capability validation exhaustive — python/pyspark/sql/udf.py:329 — see inline.
• Validate the effective schema of analyze-based Arrow UDTFs — python/pyspark/sql/connect/udtf.py:176 — see inline.
• Unwrap UDTs in the row-fallback physical schema — sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:96 — see inline.
• Propagate the query CHAR/VARCHAR policy to the state server — sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:152 — see inline.
• Preserve view-bound CHAR/VARCHAR semantics during execution — sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:85 — see inline.

Non-blocking (P2)

• Avoid eager active-session parsing for ordinary Connect UDTFs — python/pyspark/sql/connect/udtf.py:179 — see inline.
• Close local Arrow resources when assignment validation fails — sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:595 — see inline.
• Avoid checking batched UDF results twice — sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvalPythonEvaluatorFactory.scala:45 — see inline.
• Exercise every newly admitted scalar Arrow protocol — python/pyspark/sql/udf.py:323 — see inline.

Re-review status

Prior AI findings: 5 addressed, 0 still present; additional unresolved findings in this review: 9.

New attribution: 2 newly introduced, 7 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

No prior AI findings remain.

Existing discussions

• existing discussion — The requested mapper audit uncovered omitted eval types, an analyze-derived Arrow UDTF gap, and a Connect lifecycle regression.

PR metadata suggestions

• Document that Arrow UDTFs currently reject explicit CHAR/VARCHAR output schemas, including nested constrained types, because that user-visible limitation is part of this patch but is absent from the PR description.

[srielau]

Thanks for the questions.

No new configuration is needed, and this PR does not add one. CHAR/VARCHAR previously failed at these Arrow boundaries because the mapper rejected them, not because the semantics were undefined. Existing configs already own that: spark.sql.legacy.charVarcharAsString, spark.sql.preserveCharVarcharTypeInfo, and spark.sql.charVarchar.standardSemantics.enabled. Assignment uses CharVarcharUtils.shouldApplyWriteSideLengthCheck, so first-class / standard mode still pads CHAR and rejects over-length VARCHAR, and pure legacy-as-string still skips those checks.

We audited to_arrow_type / to_arrow_schema call sites. The mapper itself only normalizes CHAR/VARCHAR to physical Arrow STRING; it does not encode assignment policy. Each consumer then either stays unsupported or applies write-side checks on the JVM ingest path:

• Admitted, with JVM assignment checks: scalar batched / Arrow / pandas / pandas-iter UDFs, plus createDataFrame / toArrow via ArrowConverters.
• Explicitly rejected, including nested CHAR/VARCHAR and analyze-derived schemas: Arrow UDTFs (SQL_ARROW_UDTF / SQL_ARROW_TABLE_UDF) on both classic and Connect. ArrowEvalPythonUDTFExec still has no physical-STRING + write-side consumer.
• Default-deny for other Python eval types (grouped / map / cogrouped / window / incremental / TransformWithState): CHAR/VARCHAR return types raise PySparkNotImplementedError.

Python DataSource is the remaining gap from that audit. plan_data_source_read.py now succeeds at to_arrow_schema for CHAR/VARCHAR (physical STRING), but the JVM reader still goes through MapInBatchEvaluatorFactory, which projects identity and does not call stringLengthCheck. I will follow up by rejecting CHAR/VARCHAR on that path until it has the same assignment consumer as scalar Arrow UDFs, rather than letting the mapper change silently admit them.

[cloud-fan]

Review summary

The deeper issue is an undefined support boundary. Mapping logical CHAR/VARCHAR to physical STRING in shared conversion code broadens multiple Python consumers, while assignment enforcement and the captured policy remain consumer-specific. Before adding more per-call-site machinery, please state which surfaces this PR intends to support. My recommendation is to keep scalar Python UDFs, higher-order-function UDFs, and explicit-schema Arrow DataFrame creation in scope, and restore deterministic early rejection for Python DataSource, CHAR/VARCHAR hidden inside UDT storage, Python state schemas, and row Python UDTF result schemas. Unsupported must mean an explicit recursive validation error, not unchecked values.

Under that narrower boundary, the must-fix set is much smaller: preserve the captured policy through nested fusion and lambda lifting, handle map-key collisions after CHAR normalization, normalize legacy Arrow relation schemas to STRING, and repair the columnar, RDD-branch, and mixed-policy regressions. The predicate comment can be updated with the fusion fix. The grouping-key schema cache is a non-blocking optimization that can be deferred. The authoritative Build check was still pending in the pinned snapshot, and no tests were run during this static review.

Findings

13 total: 0 P0, 5 P1, 6 P2, 2 P3.

Blocking (P1)

• Use the child UDF's captured policy when deciding whether to fuse — sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala:204 — see inline.

• Preserve CHAR/VARCHAR checks when lifting Python UDFs — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:341 — see inline.

• Keep Python DataSource behind an explicit CHAR/VARCHAR capability gate — python/pyspark/sql/pandas/types.py:137
  Python DataSource is being enabled as a side effect of the shared Arrow mapping, not by a complete consumer design. Please keep it outside this PR's support matrix: reject recursive CHAR/VARCHAR in DataSource return schemas before to_arrow_schema and add a rejection test. That preserves the old capability boundary while scalar UDF and explicit-schema Arrow consumers remain enabled.

  Recommended change: Add a Python DataSource-specific recursive capability gate now; enable the types later only with an end-to-end query-bound JVM assignment check.

  Why this works: Validate the declared return schema before shared Arrow conversion and reject any nested CHAR/VARCHAR shape for this consumer.

  Scope: Python DataSource planning validation and focused rejection coverage.

  Compatibility: The path was previously rejected by Arrow conversion, so an explicit consumer-local rejection preserves the old supported capability while improving failure locality.

  Risks: Placing the rejection in the shared mapper would also block consumers that this PR intentionally enables. A shallow check would miss constrained strings inside arrays, maps, structs, or UDT storage.

  Constraints: Keep scalar UDF and explicit-schema Arrow support enabled. Use recursive logical-type detection at the DataSource boundary.

  Success: Python DataSource fails explicitly for every recursive CHAR/VARCHAR return shape, while supported consumers continue to map those types to Arrow STRING.

• Deduplicate map keys after CHAR/VARCHAR normalization — sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:280 — see inline.

• Normalize the schema in the legacy Arrow createDataFrame path — sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:563 — see inline.

Non-blocking (P2)

• Keep UDT-contained CHAR/VARCHAR returns behind a capability gate — sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:94
  The common mapper has implicitly enabled CHAR/VARCHAR inside UDT storage even though the evaluator's assignment checks do not understand that wrapper. For a narrower, reviewable scope, please reject any UDT whose recursive sqlType contains CHAR/VARCHAR at scalar-UDF return validation and add a rejection test. Direct and ordinary complex CHAR/VARCHAR scalar results can remain supported; UDT-contained support can be added later with one shared normalization contract.

  Recommended change: Reject recursive CHAR/VARCHAR inside UserDefinedType storage for this PR instead of adding partial evaluator support.

  Why this works: At scalar Python UDF return-type validation, distinguish UDT wrappers and fail before Arrow schema conversion when their recursive sqlType contains CHAR or VARCHAR.

  Scope: Scalar Python UDF return-type capability validation and focused UDT rejection coverage.

  Compatibility: This shape failed before the shared Arrow mapper was broadened, so an explicit gate preserves the previous supported boundary while improving the error.

  Risks: A shallow UDT check would miss nested constrained types inside its storage schema. Putting the gate in the generic mapper would incorrectly disable direct CHAR/VARCHAR support for intended consumers.

  Constraints: Keep direct struct, array, and map constrained-string scalar results supported. Reject before Python execution or Arrow conversion.

  Success: UDT-contained CHAR/VARCHAR returns fail deterministically during validation, while the supported direct and ordinary complex scalar-UDF cases retain standard and legacy behavior.

• Collect the same Arrow-columnar plan that the tests inspect — sql/core/src/test/scala/org/apache/spark/sql/execution/python/ArrowColumnarPythonUDFSuite.scala:140 — see inline.

• Keep Python state CHAR/VARCHAR schemas unsupported in this PR — sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServer.scala:93
  Python state support introduces a durable boundary and several independently mutating request families. I recommend keeping it out of this PR: reject recursive CHAR/VARCHAR in Python state and grouping-key schemas during registration or planning, remove the incidental runtime support, and test the rejection. That preserves a clear boundary without requiring the broad positive and negative state matrix identified here.

  Recommended change: Restore Python state CHAR/VARCHAR as an explicitly unsupported schema capability for this PR.

  Why this works: Validate state-variable and grouping-key schemas recursively before starting the Python state runner, and reject constrained-string types before any state request can be issued.

  Scope: Python TransformWithState schema validation, incidental state-server policy plumbing, and rejection tests.

  Compatibility: Pre-patch Python state conversion did not support these constrained values, so the gate preserves the prior functional boundary and avoids creating an unverified durable-state format.

  Risks: Validating only value state would leave list, map, grouping-key, or analyze-derived schemas exposed. Rejecting after state initialization could leave partial runtime state or confusing failures.

  Constraints: Apply the gate before starting state processing. Cover every state-variable family and recursive schema wrapper through the shared validation path.

  Success: All Python state and grouping-key schemas containing recursive CHAR/VARCHAR fail deterministically before execution, and no state mutation path depends on the new assignment policy.

• Keep row Python UDTF CHAR/VARCHAR results unsupported in this PR — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:500
  Row UDTFs are another public result protocol, including an independent analyze-derived resolution path. I recommend keeping CHAR/VARCHAR results unsupported here: reject them recursively for both fixed and analyze-returned row UDTF schemas and add focused rejection coverage. This avoids carrying partially verified policy state through UDTF planning and execution; full row-UDTF semantics can be a follow-up.

  Recommended change: Reject recursive CHAR/VARCHAR result schemas for row Python UDTFs in this PR.

  Why this works: Apply the same explicit capability validation to fixed and analyze-derived UDTF schemas before constructing an executable PythonUDTF.

  Scope: Classic row Python UDTF return-schema validation, captured-policy plumbing made unnecessary by the gate, and rejection tests.

  Compatibility: These result schemas were not successfully supported before the converter changes, so early rejection preserves the existing capability boundary.

  Risks: Checking only declared schemas would leave analyze-derived schemas admitted. Sharing the Arrow-only rejection condition without broadening it to row UDTFs would retain inconsistent behavior.

  Constraints: Validate both fixed and analyze-derived return schemas. Do not affect ordinary STRING UDTF results.

  Success: Every row UDTF return schema containing recursive CHAR/VARCHAR fails at validation with consistent behavior across declared and analyze-derived forms.

• Exercise standard checks on the RDD Arrow conversion branch — sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:584 — see inline.

• Cover heterogeneous policies in one batched UDF result — sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala:112 — see inline.

Nit (P3)

• Update the fusion contract with the policy fix — sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala:204
  Please update the exhaustive false-condition comment when changing the fusion predicate to use the child's captured policy. I view this as part of that correction, not an independent blocker.
• Defer grouping-key schema caching — sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServer.scala:361
  This repeated schema reconstruction is real but non-blocking. With the recommended narrow scope, the incidental Python state support should be removed and this concern disappears. If state support remains, caching can be a follow-up rather than a prerequisite for this PR.

Re-review status

Prior AI findings: 9 addressed, 0 still present; additional unresolved findings in this review: 13.

New attribution: 5 newly introduced, 8 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

No prior AI findings remain.

Existing discussions

• existing discussion — The author's own call-site audit confirms that this head newly admits Python DataSource without a JVM assignment consumer; promising a later rejection does not make the current defect inactive.
• existing discussion — The test moved to the requested fixture, but its padding assertion collects a new pruned DataFrame rather than the plan whose Arrow-backed child was inspected.
• existing discussion — The companion test exists, but result.select(...).collect() optimizes and executes a different, pruned plan from the one whose child was asserted columnar.
• existing discussion — Evaluator projections now honor the captured flag, but nested-UDF extraction still re-reads the later SQLConf and can remove the intermediate checked boundary.
• existing discussion — The requested shared-mapper call-site audit identifies Python DataSource as an active newly admitted path without normalization plus assignment enforcement.

PR metadata suggestions

• Add an explicit support matrix to the PR description. Under the recommended narrow scope, list scalar and higher-order Python UDFs plus explicit-schema Arrow creation as supported, and Python DataSource, UDT-contained constrained strings, Python state schemas, and row Python UDTF schemas as explicitly rejected.
• Update the user-facing and testing claims after the boundary is implemented: unsupported consumers should promise deterministic validation errors, supported legacy Arrow relations should expose STRING, and the test section should name the columnar, RDD-branch, map-collision, policy-transition, and mixed-policy regressions.

[cloud-fan]

Non-blocking (P2): This inspects padded but collects padded.select("udf_id"), which is a separately optimized query. The added projection can prune the Arrow source and make the collected query fall back to row input, so the test can pass without exercising the asserted columnar path. Please collect padded itself, read the UDF field at its original ordinal, and make the same change in the legacy case.

[cloud-fan]

Blocking (P1): This decision re-reads the querying session's SQLConf even though the child UDF now carries its resolved policy. If a view is resolved under standard semantics and queried from a legacy session, this can fuse away the child's JVM boundary, so the outer UDF sees an unpadded CHAR or an over-length VARCHAR avoids its error. Please gate on child.applyCharVarcharChecks && hasCharVarchar(child.dataType) and add a construction-versus-execution config-transition regression.

Recommended change: Make nested-UDF fusion depend on the resolved child expression's captured policy rather than optimizer-time SQLConf.

Why this works: Replace the ambient write-side policy read with child.applyCharVarcharChecks while retaining the constrained-type predicate, so checked intermediates keep a JVM conversion boundary.

Scope: ExtractPythonUDFs nested-chain extraction and a persisted or lazy-plan configuration-transition test.

Compatibility: Preserve current fusion for unchecked children and unconstrained result types; only prevent fusion where the resolved child already requires assignment checks.

Risks: Blocking fusion adds a Python evaluation boundary for constrained checked intermediates. A regression test that does not separate resolution from execution would miss the lifecycle defect.

Constraints: Do not re-resolve policy from the current SQLConf. Keep the change local to constrained child results whose captured flag is true.

Success: The same resolved nested-UDF plan returns identical checked results when later queried under an opposite CHAR/VARCHAR session policy.

[cloud-fan]

Blocking (P1): The lambda rewrite builds a replacement PythonUDF but does not copy this newly captured flag, so its default becomes false. A CHAR/VARCHAR UDF inside transform can then return an unpadded CHAR or accept an over-length VARCHAR. Please pass applyCharVarcharChecks = udf.applyCharVarcharChecks when constructing the lifted UDF and add higher-order-function coverage for both outcomes.

[cloud-fan]

Blocking (P1): Key conversion can now coalesce distinct Python keys: for MapType(CharType(2), ...), both "a" and "a " become the same Catalyst key. Passing those entries directly to ArrayBasedMapData violates its documented no-duplicates precondition. Please build through the duplicate-aware map builder, or otherwise detect post-normalization collisions and honor spark.sql.mapKeyDedupPolicy.

Recommended change: Route converted map entries through Spark's duplicate-aware map construction policy after key normalization.

Why this works: Normalize each key first, then use ArrayBasedMapBuilder or equivalent collision detection so equal normalized keys follow spark.sql.mapKeyDedupPolicy.

Scope: Python-to-Catalyst map conversion for constrained key types and collision regression tests.

Compatibility: Preserve behavior for maps whose normalized keys remain unique; make newly created collisions follow the same policy as other Catalyst map constructors.

Risks: Applying deduplication before CHAR normalization would miss the collision. Ignoring the configured LAST_WIN policy or exception behavior would diverge from Spark SQL map semantics.

Constraints: Do not admit duplicate keys into ArrayBasedMapData. Apply the configured duplicate-key policy to the normalized Catalyst keys.

Success: Keys such as a and a either raise the configured duplicate-key error or deterministically deduplicate under LAST_WIN, and all unique maps remain unchanged.

[cloud-fan]

Blocking (P1): In legacy-as-string mode this flag is false, but attrs and the schema passed to both relation branches still contain CHAR/VARCHAR while first-class types are disabled. CheckAnalysis rejects those leaf outputs before the intended unchecked values can be read. Please normalize the relation schema and attributes to STRING under the existing legacy policy, retaining the declared logical schema only where standard-mode projection needs it.

Recommended change: Separate the declared schema used for standard assignment checks from the policy-normalized schema exposed by Arrow relations.

Why this works: Under legacy-as-string mode, derive relation attributes and the LogicalRDD or LocalRelation schema from the existing CHAR/VARCHAR-to-STRING normalization while keeping the original schema only for Arrow decoding or standard checks where required.

Scope: ArrowConverters.toDataFrame relation construction and any Python-side schema caching that exposes the unnormalized logical schema.

Compatibility: Restore the established legacy contract that constrained strings appear as STRING and remain unchecked; preserve first-class logical types under standard semantics.

Risks: Normalizing the decoding schema too early could lose information needed for standard-mode checks. Using different attribute sequences for projection and relation construction can cause expression-ID or type mismatches.

Constraints: Both the RDD and local relation branches must expose the same policy-normalized schema. Do not normalize away CHAR/VARCHAR in standard semantics.

Success: Both Arrow relation branches analyze and return unchecked STRING values in legacy mode, while standard mode still pads CHAR and rejects over-length VARCHAR.

[cloud-fan]

Non-blocking (P2): This is a separately implemented executor-side projection, but the standard-semantics tests all stay on the local branch; the only test forcing threshold 0 enables legacy mode, where this block is skipped. Please force the RDD branch under standard semantics and assert both CHAR padding and an EXCEED_LIMIT_LENGTH failure during materialization.

[cloud-fan]

Non-blocking (P2): This sequence is consumed positionally, but current multi-UDF coverage makes every flag false. Please construct two independent expressions under opposite policies, evaluate them together in one BatchEvalPythonExec, and assert that the checked CHAR field is padded while the unchecked over-length VARCHAR field is preserved at the matching ordinal.