[FLINK-39700][table-planner] Fix column-order sensitivity in validateAndExtractColumnChanges

[AHeise]

What is the purpose of the change

Physical DDL columns in CREATE OR ALTER MATERIALIZED TABLE could not be declared in a different order than the AS SELECT projection, because validateAndExtractColumnChanges compared old and new schemas positionally. This PR switches to name-based matching so that type changes, computed-column definition changes, drops, and additions each generate the appropriate TableChange entry.

It also consolidates query-change and persisted-column-drop validation — previously scattered across MaterializedTableChangeHandler and SqlAlterMaterializedTableDropSchemaConverter — into AlterMaterializedTableChangeOperation.validateChanges(). The handler becomes a pure applier.

CREATE OR ALTER now uses declarative semantics: non-persisted columns absent from the explicit DDL are dropped instead of kept.

Brief change log

• MaterializedTableUtils.validateAndExtractColumnChanges: replace positional comparison with name-based lookup over all column kinds (physical, computed, metadata):
  • Physical type change → ModifyPhysicalColumnType (instead of throwing)
  • Computed/metadata definition change → ModifyColumn
  • Comment-only change → ModifyColumnComment (also emitted alongside ModifyPhysicalColumnType when both change)
  • Column absent in new schema → DropColumn; non-persisted columns are skipped when schemaDefinedInQuery=false (schema derived from query projection, so absent non-persisted columns are assumed to still exist rather than being intentionally removed)
  • New column → AddColumn; nullable coercion applied when schemaDefinedInQuery=false
  • Removes the apache-commons-lang3 StringUtils dependency from this class
• AlterMaterializedTableChangeOperation: new validateChanges() method called from execute() rejects:
  • Column reordering (ModifyColumnPosition) when the change set includes ModifyDefinitionQuery
  • Physical type changes (ModifyPhysicalColumnType) when the change set includes ModifyDefinitionQuery
  • Persisted-column drops (DropColumn) unconditionally
    Also gains computeNewTable() (protected hook), getOldTable(), and setOldTable() (invalidates derived caches)
• MaterializedTableChangeHandler: remove isQueryChange, droppedPersistedCnt, validationErrors, checkForChangedPositionByQuery, and all positionChangeError helpers; dropColumn is now unconditional; unknown changes throw immediately
• SqlAlterMaterializedTableDropSchemaConverter: remove the per-column persisted-column-drop guard (now handled by validateChanges)
• FullAlterMaterializedTableOperation: override computeNewTable() with a newTableBuilder lambda supplied by SqlCreateOrAlterMaterializedTableConverter, so the caller controls the new-table construction without shadowing the cached newTable field on the parent
• SqlCreateOrAlterMaterializedTableConverter: extract buildNewTable() that builds the new table definition with non-persisted columns absent from the explicit DDL dropped (declarative semantics); unfold buildTableChanges from a returned lambda into a direct method; add getMergedLogicalRefreshMode, getMergedComment, getMergedFreshness to MergeContext

Verifying this change

This change added tests and can be verified as follows:

• AlterMaterializedTableChangeOperationValidationTest: rejects position changes, type changes, and persisted-column drops via validateChanges()
• AlterMaterializedTableAsQueryOperationValidationTest: end-to-end validation of CREATE OR ALTER and ALTER ... AS SELECT through execute()
• MaterializedTableChangeHandlerTest: handler applies changes without validation side effects
• ValidateAndExtractColumnChangesTest: parametrized suite covering identical schemas, comment add/remove, column appended (with/without schemaDefinedInQuery), reordered persisted columns (no-op), type change, drop, rename, computed add/drop/modify, metadata add/drop
• SqlMaterializedTableNodeToOperationConverterTest: updated expected error messages and schema assertions

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no (correctness fix and refactor for materialized table schema evolution)
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (Claude Code)

Generated-by: Claude Code (claude-sonnet-4-6)

[flinkbot]

CI report:

• f750378 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[AHeise]

@flinkbot run azure

[AHeise]

@flinkbot run azure

[AHeise]

@flinkbot run azure

[snuyanzin]

nit

Suggested change

	final Integer idx = columnIndex.get(change.getColumnName());
	final int idx = columnIndex.getOrDefault(change.getColumnName(), -1);

and then have logic similar to checkPositionChange
with

if (index < 0) {
            return;
        }
        errors.add(
...

[snuyanzin]

same for checkPhysicalTypeChange

[AHeise]

Fixed.

[AHeise]

Fixed.

[snuyanzin]

WDYT about movement public methods up, private to the bottom?

[AHeise]

Done, moved public methods up and private to the bottom.

[AHeise]

@flinkbot run azure

[snuyanzin]

LGTM
Thanks for addressing feedback