This is an automated email from the ASF dual-hosted git repository.
AHeise pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7d194385601 [FLINK-40496][table-planner] Apply append-only column
rules consistently when altering a materialized table (#29034)
7d194385601 is described below
commit 7d194385601701283df5da9f296f3eb5c61b46ba
Author: Arvid Heise <[email protected]>
AuthorDate: Wed Sep 2 16:11:00 2026 +0200
[FLINK-40496][table-planner] Apply append-only column rules consistently
when altering a materialized table (#29034)
CREATE OR ALTER derived its column changes from a different diff
implementation than ALTER ... AS and did not surface a query-driven column
reorder to the append-only validation, so reordering existing columns behaved
inconsistently: it was silently applied when the query text was unchanged -
rewriting the stored column order with no error - and rejected otherwise. The
silent path left the stored schema disagreeing with the query, which then
miscompiled the positional refresh INSERT.
Route both statements through one diff (validateAndExtractColumnChanges,
dropping buildSchemaTableChanges), computing the CREATE OR ALTER diff from the
query the same way ALTER ... AS does and positioning old columns by their rank
among the columns that survive into the new schema so retained non-persisted
columns do not skew it. Apply the append-only rules to every query-carrying
alter regardless of whether the query text changed, so reordering or retyping
existing columns is rejecte [...]
Co-authored-by: Arvid Heise <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
.../service/MaterializedTableStatementITCase.java | 79 ++++----
.../AlterMaterializedTableChangeOperation.java | 5 +-
...ializedTableAsQueryOperationValidationTest.java | 31 ++++
.../SqlAlterMaterializedTableAsQueryConverter.java | 3 +-
.../planner/utils/MaterializedTableUtils.java | 198 +++++----------------
...erializedTableNodeToOperationConverterTest.java | 7 -
...reateOrAlterMaterializedTableConverterTest.java | 31 ++++
.../planner/utils/MaterializedTableUtilsTest.java | 108 -----------
.../utils/ValidateAndExtractColumnChangesTest.java | 60 ++++++-
9 files changed, 210 insertions(+), 312 deletions(-)
diff --git
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
index ddc0fcd8500..f71337cfd11 100644
---
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
+++
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
@@ -61,6 +61,8 @@ import org.apache.flink.types.Row;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Trigger;
@@ -1082,19 +1084,27 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
.containsExactly("shop_id", "user_id", "oca", "order_id");
}
- @Test
- void testCreateOrAlterMaterializedTableColumnListIgnoredOnAlterPath()
throws Exception {
- createAndVerifyCreateMaterializedTableWithData(
- "users_shops", List.of(), Map.of(), RefreshMode.FULL);
-
- ObjectIdentifier userShopsIdentifier =
getObjectIdentifier("users_shops");
- ResolvedSchema oldSchema =
getTable(userShopsIdentifier).getResolvedSchema();
- assertThat(oldSchema.getColumnNames())
- .containsExactly("user_id", "shop_id", "ds", "order_cnt");
-
- // users_shops already exists as a materialized table, so CREATE OR
ALTER takes the alter
- // path. The DDL column list is treated as names only: it does NOT
reorder the schema.
- String materializedTableDDL =
+ // Swaps the first two projections so existing columns are reordered
rather than appended.
+ private static final String REORDERED_QUERY =
+ " AS SELECT \n"
+ + " shop_id,\n"
+ + " user_id,\n"
+ + " ds,\n"
+ + " COUNT(order_id) AS order_cnt\n"
+ + " FROM (\n"
+ + " SELECT user_id, shop_id, order_created_at AS ds,
order_id FROM my_source"
+ + " ) AS tmp\n"
+ + " GROUP BY (user_id, shop_id, ds)";
+
+ /**
+ * Reordering existing columns of a materialized table is rejected on
every query-carrying alter
+ * path, and the rejected statement leaves the stored schema untouched.
+ */
+ @ParameterizedTest(name = "[{index}]")
+ @ValueSource(
+ strings = {
+ // A bare column list orders columns on CREATE, so CREATE OR
ALTER honors it on the
+ // alter path: listing the existing columns in a different
order is a reorder.
"CREATE OR ALTER MATERIALIZED TABLE users_shops (shop_id,
user_id, ds, order_cnt)"
+ " PARTITIONED BY (ds)\n"
+ " WITH(\n"
@@ -1108,47 +1118,28 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
+ " FROM (\n"
+ " SELECT user_id, shop_id, order_created_at AS
ds, order_id FROM my_source"
+ " ) AS tmp\n"
- + " GROUP BY (user_id, shop_id, ds)";
- OperationHandle handle = executeStatement(materializedTableDDL);
- awaitOperationTermination(service, sessionHandle, handle);
-
- ResolvedSchema newSchema =
getTable(userShopsIdentifier).getResolvedSchema();
- assertThat(newSchema.getColumnNames())
- .containsExactly("user_id", "shop_id", "ds", "order_cnt");
- assertThat(newSchema).isEqualTo(oldSchema);
- }
-
- @Test
- void testAlterMaterializedTableAsQueryRejectsReorder() throws Exception {
+ + " GROUP BY (user_id, shop_id, ds)",
+ // ALTER ... AS reorders existing columns through the query
projection.
+ "ALTER MATERIALIZED TABLE users_shops" + REORDERED_QUERY,
+ // CREATE OR ALTER ... AS reorders existing columns through
the query projection.
+ "CREATE OR ALTER MATERIALIZED TABLE users_shops" +
REORDERED_QUERY
+ })
+ void rejectsReorderingExistingColumns(String statement) throws Exception {
createAndVerifyCreateMaterializedTableWithData(
"users_shops", List.of(), Map.of(), RefreshMode.FULL);
ObjectIdentifier userShopsIdentifier =
getObjectIdentifier("users_shops");
-
assertThat(getTable(userShopsIdentifier).getResolvedSchema().getColumnNames())
+ ResolvedSchema oldSchema =
getTable(userShopsIdentifier).getResolvedSchema();
+ assertThat(oldSchema.getColumnNames())
.containsExactly("user_id", "shop_id", "ds", "order_cnt");
- // ALTER ... AS re-derives the schema from the new query. Swapping the
first two
- // projections asks to reorder existing columns. The query-evolution
rules reject this: an
- // existing materialized table may only append columns at the end, not
reorder them.
- String alterMaterializedTableAsQueryDDL =
- "ALTER MATERIALIZED TABLE users_shops"
- + " AS SELECT \n"
- + " shop_id,\n"
- + " user_id,\n"
- + " ds,\n"
- + " COUNT(order_id) AS order_cnt\n"
- + " FROM (\n"
- + " SELECT user_id, shop_id, order_created_at AS
ds, order_id FROM my_source"
- + " ) AS tmp\n"
- + " GROUP BY (user_id, shop_id, ds)";
- OperationHandle handle =
executeStatement(alterMaterializedTableAsQueryDDL);
+ OperationHandle handle = executeStatement(statement);
assertThatThrownBy(() -> awaitOperationTermination(service,
sessionHandle, handle))
.hasStackTraceContaining("reordering columns are not
supported");
- // The rejected alter leaves the original schema untouched.
-
assertThat(getTable(userShopsIdentifier).getResolvedSchema().getColumnNames())
- .containsExactly("user_id", "shop_id", "ds", "order_cnt");
+ // The rejected statement leaves the original schema untouched.
+
assertThat(getTable(userShopsIdentifier).getResolvedSchema()).isEqualTo(oldSchema);
}
@Test
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java
index 299c72d6cfe..3de92ec1702 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java
@@ -120,8 +120,11 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
@VisibleForTesting
public void validateChanges() {
final List<TableChange> changes = getTableChanges();
+ // CoA and ALTER ... AS carry the defining query, so the append-only
column rules apply even
+ // when the query text is unchanged; metadata-only DDL alters carry no
query.
final boolean isQueryChange =
-
changes.stream().anyMatch(ModifyDefinitionQuery.class::isInstance);
+ asQueryOperation != null
+ ||
changes.stream().anyMatch(ModifyDefinitionQuery.class::isInstance);
final List<Column> oldColumns =
oldTable.getResolvedSchema().getColumns();
final Map<String, Integer> columnIndex =
IntStream.range(0, oldColumns.size())
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
index 88073daca8b..47a0b5c1b6d 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
@@ -163,6 +163,37 @@ class AlterMaterializedTableAsQueryOperationValidationTest
{
"Column mismatch at position 1: Original column is
[`a` INT], but new column is [`a` BIGINT].");
}
+ @Test
+ void rejectInsertColumnBeforeMultipleColumns() {
+ // (a, b, c) -> (a, mid, b, c): both b and c shift; c is repositioned
after the existing
+ // column b, which the append-only guard rejects.
+ final ResolvedCatalogMaterializedTable oldTable =
+ resolvedTable(
+ ResolvedSchema.of(
+ physical("a", DataTypes.INT()),
+ physical("b", DataTypes.STRING()),
+ physical("c", DataTypes.BIGINT())));
+
+ final AlterMaterializedTableAsQueryOperation op =
+ operation(
+ oldTable,
+ List.of(
+ TableChange.modifyDefinitionQuery(
+ "SELECT a, 1 AS mid, b, c FROM src",
+ "SELECT `src`.`a`, 1 AS `mid`,
`src`.`b`, `src`.`c` FROM `src`"),
+ TableChange.add(physical("mid",
DataTypes.INT())),
+ TableChange.modifyColumnPosition(
+ physical("b", DataTypes.STRING()),
+ ColumnPosition.after("mid")),
+ TableChange.modifyColumnPosition(
+ physical("c", DataTypes.BIGINT()),
+ ColumnPosition.after("b"))));
+
+ assertThatThrownBy(op::validateChanges)
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Column mismatch at position 3");
+ }
+
@Test
void acceptAppendColumn() {
final ResolvedCatalogMaterializedTable oldTable =
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
index eb631523b98..e6f922984bc 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
@@ -74,7 +74,8 @@ public class SqlAlterMaterializedTableAsQueryConverter
ResolvedSchema newSchema = queryOperation.getResolvedSchema();
List<TableChange> tableChanges =
new ArrayList<>(
-
MaterializedTableUtils.buildSchemaTableChanges(oldSchema, newSchema));
+
MaterializedTableUtils.validateAndExtractColumnChanges(
+ oldSchema, newSchema, false));
if (!tableChanges.isEmpty()) {
final boolean hasNonPersistedColumn =
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
index 69c7277ab1a..2fdc32a5dbd 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
@@ -262,91 +262,6 @@ public class MaterializedTableUtils {
}
}
- // Used to build changes introduced by changed query like
- // ALTER MATERIALIZED TABLE ... AS ...
- public static List<TableChange> buildSchemaTableChanges(
- ResolvedSchema oldSchema, ResolvedSchema newSchema) {
- if (!isSchemaChanged(oldSchema, newSchema)) {
- return List.of();
- }
-
- final List<Column> oldColumns = oldSchema.getColumns();
- final Map<String, Tuple2<Column, Integer>> oldColumnSet = new
HashMap<>();
- for (int i = 0; i < oldColumns.size(); i++) {
- Column column = oldColumns.get(i);
- oldColumnSet.put(column.getName(), Tuple2.of(oldColumns.get(i),
i));
- }
- // Schema retrieved from query doesn't count existing non persisted
columns
- final List<Column> newColumns = newSchema.getColumns();
-
- List<TableChange> changes = new ArrayList<>();
- for (int i = 0; i < newColumns.size(); i++) {
- Column newColumn = newColumns.get(i);
- Tuple2<Column, Integer> oldColumnToPosition =
oldColumnSet.get(newColumn.getName());
-
- if (oldColumnToPosition == null) {
-
changes.add(TableChange.add(newColumn.copy(newColumn.getDataType().nullable())));
- continue;
- }
-
- // Check if position changed
- applyPositionChanges(newColumns, oldColumnToPosition, i, changes);
-
- Column oldColumn = oldColumnToPosition.f0;
- // Check if column changed
- // Note: it could be unchanged while the position is changed
- if (oldColumn.equals(newColumn)) {
- // no changes
- continue;
- }
-
- // Check if kind changed
- if (oldColumn.getClass() != newColumn.getClass()) {
- changes.add(TableChange.dropColumn(oldColumn.getName()));
-
changes.add(TableChange.add(newColumn.copy(newColumn.getDataType().nullable())));
- continue;
- }
-
- // Check if comment is changed
- if (!Objects.equals(
- oldColumn.getComment().orElse(null),
newColumn.getComment().orElse(null))) {
- changes.add(
- TableChange.modifyColumnComment(
- oldColumn,
newColumn.getComment().orElse(null)));
- }
-
- // Check if physical column type changed
- if (oldColumn.isPhysical()
- && newColumn.isPhysical()
- &&
!oldColumn.getDataType().equals(newColumn.getDataType())) {
- changes.add(
- TableChange.modifyPhysicalColumnType(oldColumn,
newColumn.getDataType()));
- }
-
- // Check if metadata fields changed
- if (oldColumn instanceof MetadataColumn) {
- applyMetadataColumnChanges(
- (MetadataColumn) oldColumn, (MetadataColumn)
newColumn, changes);
- }
-
- // Check if computed expression changed
- if (oldColumn instanceof ComputedColumn) {
- applyComputedColumnChanges(
- (ComputedColumn) oldColumn, (ComputedColumn)
newColumn, changes);
- }
- }
-
- for (Column newColumn : newColumns) {
- oldColumnSet.remove(newColumn.getName());
- }
-
- for (Map.Entry<String, Tuple2<Column, Integer>> entry :
oldColumnSet.entrySet()) {
- changes.add(TableChange.dropColumn(entry.getKey()));
- }
-
- return changes;
- }
-
private static boolean isDateTimeInterval(SqlTypeName typeName) {
return typeName == SqlTypeName.INTERVAL_DAY
|| typeName == SqlTypeName.INTERVAL_HOUR
@@ -374,33 +289,6 @@ public class MaterializedTableUtils {
}
}
- // Since it is only for query change, then check only persisted columns
which could be
- // changed/added/dropped with such change
- private static boolean isSchemaChanged(ResolvedSchema oldSchema,
ResolvedSchema newSchema) {
- List<Column> oldPersistedColumns =
- oldSchema.getColumns().stream()
- .filter(Column::isPersisted)
- .collect(Collectors.toList());
- if (oldPersistedColumns.size() != newSchema.getColumnCount()) {
- return true;
- }
- for (int i = 0; i < oldPersistedColumns.size(); i++) {
- Column oldColumn = oldPersistedColumns.get(i);
- Column newColumn = newSchema.getColumn(i).get();
- if (!oldColumn.getName().equals(newColumn.getName())) {
- return true;
- }
- if (!newColumn
- .getDataType()
- .getLogicalType()
- .equals(oldColumn.getDataType().getLogicalType())) {
- return true;
- }
- }
-
- return false;
- }
-
private static void applyPositionChanges(
List<Column> newColumns,
Tuple2<Column, Integer> oldColumnToPosition,
@@ -417,42 +305,46 @@ public class MaterializedTableUtils {
}
}
- private static void applyComputedColumnChanges(
- ComputedColumn oldColumn, ComputedColumn newColumn,
List<TableChange> changes) {
- if (!oldColumn
- .getExpression()
- .asSerializableString()
-
.equals(newColumn.getExpression().asSerializableString())
- && !Objects.equals(
- oldColumn.explainExtras().orElse(null),
- newColumn.explainExtras().orElse(null))) {
- // for now there is no dedicated table change
- changes.add(TableChange.dropColumn(oldColumn.getName()));
-
changes.add(TableChange.add(newColumn.copy(newColumn.getDataType().nullable())));
- }
- }
-
- private static void applyMetadataColumnChanges(
- MetadataColumn oldColumn, MetadataColumn newColumn,
List<TableChange> changes) {
- if (oldColumn.isVirtual() != newColumn.isVirtual()
- || !Objects.equals(
- oldColumn.getMetadataKey().orElse(null),
- newColumn.getMetadataKey().orElse(null))) {
- // for now there is no dedicated table change
- changes.add(TableChange.dropColumn(oldColumn.getName()));
-
changes.add(TableChange.add(newColumn.copy(newColumn.getDataType().nullable())));
- }
- }
-
+ /**
+ * Computes the column-level {@link TableChange}s between a materialized
table's current schema
+ * and its new schema (query-derived, or DDL-defined when {@code
schemaDefinedInQuery}). The
+ * result feeds the append-only enforcement in {@code
+ * AlterMaterializedTableChangeOperation#validateChanges}. Per column it
emits:
+ *
+ * <ul>
+ * <li>{@code add} — a column absent from the old schema (nullable when
query-derived);
+ * <li>{@code modifyColumnPosition} — an existing column the query
moved; the query order is
+ * authoritative, so a reorder surfaces here (and is later
rejected). A DDL column list
+ * keeps its own order and emits no reposition;
+ * <li>{@code modifyPhysicalColumnType} — a physical-type change. For a
query-derived schema a
+ * tightening nullability flip (nullable to NOT NULL) is tolerated
as an inference
+ * artifact, while a loosening flip is surfaced; a DDL-defined
schema surfaces any
+ * difference;
+ * <li>{@code modifyColumn} / {@code modifyColumnComment} — changed
computed/metadata
+ * definitions or comments;
+ * <li>{@code dropColumn} — an old column absent from the new schema;
old non-persisted
+ * columns are retained when the schema is query-derived.
+ * </ul>
+ *
+ * <p>Existing columns are ranked among the columns that survive into the
new schema, so
+ * retained non-persisted columns do not skew the position diff.
+ */
public static List<TableChange> validateAndExtractColumnChanges(
ResolvedSchema oldSchema, ResolvedSchema newSchema, boolean
schemaDefinedInQuery) {
final List<Column> oldColumns = oldSchema.getColumns();
+ final List<Column> newColumns = newSchema.getColumns();
+ final Set<String> newColumnNames =
+
newColumns.stream().map(Column::getName).collect(Collectors.toSet());
+ // Position each old column among the columns that survive into the
new schema, so retained
+ // non-persisted columns (absent from the query projection) do not
skew the position diff.
final Map<String, Tuple2<Column, Integer>> oldByName = new HashMap<>();
- for (int i = 0; i < oldColumns.size(); i++) {
- oldByName.put(oldColumns.get(i).getName(),
Tuple2.of(oldColumns.get(i), i));
+ int nextPosition = 0;
+ for (final Column oldColumn : oldColumns) {
+ final Integer position =
+ newColumnNames.contains(oldColumn.getName()) ?
nextPosition++ : null;
+ oldByName.put(oldColumn.getName(), Tuple2.of(oldColumn, position));
}
final Set<String> seen = new HashSet<>();
- final List<Column> newColumns = newSchema.getColumns();
final List<TableChange> changes = new ArrayList<>();
for (int newIndex = 0; newIndex < newColumns.size(); newIndex++) {
final Column newColumn = newColumns.get(newIndex);
@@ -463,8 +355,11 @@ public class MaterializedTableUtils {
continue;
}
final Column oldColumn = oldEntry.f0;
- // No position diff: DDL order is arbitrary; query-driven reorders
are caught by
- // buildSchemaTableChanges on the ALTER MT AS path.
+ // The query order is authoritative, so reposition a column the
query moved; a
+ // DDL-defined schema keeps the arbitrary DDL order.
+ if (!schemaDefinedInQuery) {
+ applyPositionChanges(newColumns, oldEntry, newIndex, changes);
+ }
if (oldColumn.isPhysical()
&& newColumn.isPhysical()
&& typeChanged(oldColumn, newColumn,
schemaDefinedInQuery)) {
@@ -541,11 +436,16 @@ public class MaterializedTableUtils {
Column oldColumn, Column newColumn, boolean schemaDefinedInQuery) {
final DataType oldType = oldColumn.getDataType();
final DataType newType = newColumn.getDataType();
- // schemaDefinedInQuery=false: schema is inferred from the query,
which may flip
- // nullability without intent — only the base type difference is a
real change.
- return schemaDefinedInQuery
- ? !oldType.equals(newType)
- : !oldType.nullable().equals(newType.nullable());
+ if (schemaDefinedInQuery) {
+ return !oldType.equals(newType);
+ }
+ // Query-inferred nullability is a real change only when it loosens
(NOT NULL -> nullable):
+ // the stored column can no longer hold the query's possible nulls. A
tightening is
+ // tolerated.
+ final boolean baseTypeChanged =
!oldType.nullable().equals(newType.nullable());
+ final boolean loosened =
+ !oldType.getLogicalType().isNullable() &&
newType.getLogicalType().isNullable();
+ return baseTypeChanged || loosened;
}
public static ResolvedSchema getQueryOperationResolvedSchema(
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
index a90a58dcd08..70f0cf3e106 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
@@ -849,13 +849,6 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ "renaming, and reordering columns are not
supported.\n"
+ "Column mismatch at position 4: Original
column is [`d` STRING], "
+ "but new column is [`d` INT]."),
- TestSpec.of(
- "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c,
CAST('d' AS STRING) AS d FROM t3",
- "When modifying the query of a materialized table,
currently only support "
- + "appending columns at the end of original
schema, dropping, "
- + "renaming, and reordering columns are not
supported.\n"
- + "Column mismatch at position 4: Original
column is [`d` STRING], "
- + "but new column is [`d` STRING NOT NULL]."),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl_with_non_persisted
AS SELECT '123'",
"ALTER query for MATERIALIZED TABLE "
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java
index 1eeeeafc627..583c6acb55e 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java
@@ -52,6 +52,7 @@ import java.util.Map;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest
@@ -99,6 +100,36 @@ class
SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest
TableChange.reset("format"));
}
+ /**
+ * A new column inserted ahead of several trailing columns shifts every
one of them, reordering
+ * existing columns relative to each other. {@code validateChanges}
rejects that: a materialized
+ * table may only append columns, never reposition existing ones.
+ */
+ @Test
+ void testAlterMaterializedTableAsQueryRejectsInsertBeforeExistingColumns()
{
+ final String sql =
+ "CREATE OR ALTER MATERIALIZED TABLE mt AS SELECT a, 42 AS mid,
b, c, d FROM t1";
+ final FullAlterMaterializedTableOperation op =
+ (FullAlterMaterializedTableOperation) parse(sql);
+
+ assertThatThrownBy(op::validateChanges)
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Column mismatch at position 3");
+ }
+
+ /** A bare column list naming the columns in their existing order changes
nothing. */
+ @Test
+ void testAlterMaterializedTableColumnListMatchingOrderIsNoOp() {
+ final String sql = "CREATE OR ALTER MATERIALIZED TABLE mt (a, b, c, d)
AS SELECT * FROM t1";
+ final FullAlterMaterializedTableOperation op =
+ (FullAlterMaterializedTableOperation) parse(sql);
+
+ assertThatNoException().isThrownBy(op::validateChanges);
+ assertThat(op.getNewTable().getUnresolvedSchema().getColumns())
+ .map(Schema.UnresolvedColumn::getName)
+ .containsExactly("a", "b", "c", "d");
+ }
+
@Test
void testAlterMaterializedTableAsQueryWithoutDefinedSchema() {
String sql =
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/MaterializedTableUtilsTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/MaterializedTableUtilsTest.java
deleted file mode 100644
index 1859c0f531b..00000000000
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/MaterializedTableUtilsTest.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * 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.flink.table.planner.utils;
-
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.catalog.Column;
-import org.apache.flink.table.catalog.ResolvedSchema;
-import org.apache.flink.table.catalog.TableChange;
-import org.apache.flink.table.catalog.TableChange.ColumnPosition;
-
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.MethodSource;
-
-import java.util.Collection;
-import java.util.List;
-
-import static org.apache.flink.table.catalog.Column.physical;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-
-/** Tests for {@link MaterializedTableUtils}. */
-class MaterializedTableUtilsTest {
- @ParameterizedTest
- @MethodSource("input")
- void test(TestSpec spec) {
-
assertThat(MaterializedTableUtils.buildSchemaTableChanges(spec.oldSchema,
spec.newSchema))
- .isEqualTo(spec.expected);
- }
-
- private static Collection<TestSpec> input() {
- return List.of(
- TestSpec.of(
- schema(physical("a", DataTypes.INT())),
- schema(physical("a", DataTypes.INT())),
- List.of()),
- TestSpec.of(
- schema(physical("a",
DataTypes.INT()).withComment("comment")),
- schema(physical("a",
DataTypes.INT()).withComment("comment")),
- List.of()),
- TestSpec.of(
- schema(physical("a",
DataTypes.INT()).withComment("comment")),
- schema(physical("a2",
DataTypes.STRING()).withComment("comment 2")),
- List.of(
- TableChange.add(
- physical("a2", DataTypes.STRING())
- .withComment("comment 2")),
- TableChange.dropColumn("a"))),
- TestSpec.of(
- schema(physical("a", DataTypes.INT())),
- schema(physical("b", DataTypes.INT())),
- List.of(
- TableChange.add(physical("b",
DataTypes.INT())),
- TableChange.dropColumn("a"))),
- TestSpec.of(
- schema(physical("a", DataTypes.INT()), physical("b",
DataTypes.BOOLEAN())),
- schema(physical("b", DataTypes.BOOLEAN()),
physical("a", DataTypes.INT())),
- List.of(
- TableChange.modifyColumnPosition(
- physical("b", DataTypes.BOOLEAN()),
ColumnPosition.first()),
- TableChange.modifyColumnPosition(
- physical("a", DataTypes.INT()),
- ColumnPosition.after("b")))),
- TestSpec.of(
- schema(physical("a", DataTypes.INT())),
- schema(physical("a", DataTypes.BIGINT())),
- List.of(
- TableChange.modifyPhysicalColumnType(
- physical("a", DataTypes.INT()),
DataTypes.BIGINT()))));
- }
-
- private static ResolvedSchema schema(Column... columns) {
- return ResolvedSchema.of(columns);
- }
-
- private static class TestSpec {
- private final ResolvedSchema oldSchema;
- private final ResolvedSchema newSchema;
- private final List<TableChange> expected;
-
- public TestSpec(
- ResolvedSchema oldSchema, ResolvedSchema newSchema,
List<TableChange> expected) {
-
- this.oldSchema = oldSchema;
- this.newSchema = newSchema;
- this.expected = expected;
- }
-
- public static TestSpec of(
- ResolvedSchema oldSchema, ResolvedSchema newSchema,
List<TableChange> expected) {
- return new TestSpec(oldSchema, newSchema, expected);
- }
- }
-}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/ValidateAndExtractColumnChangesTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/ValidateAndExtractColumnChangesTest.java
index 3d3f177af24..7ee53c94e91 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/ValidateAndExtractColumnChangesTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/ValidateAndExtractColumnChangesTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableChange.ColumnPosition;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.utils.ResolvedExpressionMock;
import org.apache.flink.table.types.DataType;
@@ -100,10 +101,19 @@ class ValidateAndExtractColumnChangesTest {
TableChange.add(physical("b",
DataTypes.STRING())),
TableChange.add(physical("c",
DataTypes.BOOLEAN())))),
TestSpec.of(
- "nullability differs but schema is not defined in
query",
+ "loosening nullability of a query-defined column emits
modifyPhysicalColumnType",
schema(physical("a", DataTypes.INT().notNull())),
schema(physical("a", DataTypes.INT())),
false,
+ List.of(
+ TableChange.modifyPhysicalColumnType(
+ physical("a",
DataTypes.INT().notNull()),
+ DataTypes.INT()))),
+ TestSpec.of(
+ "tightening nullability inferred from a query is
tolerated",
+ schema(physical("a", DataTypes.INT())),
+ schema(physical("a", DataTypes.INT().notNull())),
+ false,
List.of()),
TestSpec.of(
"computed columns are ignored in persisted comparison",
@@ -233,7 +243,53 @@ class ValidateAndExtractColumnChangesTest {
new TableChange.ModifyColumn(
computed("comp",
expr(DataTypes.INT())),
computed("comp",
expr(DataTypes.BIGINT())),
- null))));
+ null))),
+ TestSpec.of(
+ "query inserts a column mid-projection, repositioning
the trailing column",
+ schema(
+ physical("city", DataTypes.STRING()),
+ physical("user_count", DataTypes.BIGINT())),
+ schema(
+ physical("city", DataTypes.STRING()),
+ physical("name_initial", DataTypes.STRING()),
+ physical("user_count", DataTypes.BIGINT())),
+ false,
+ List.of(
+ TableChange.add(physical("name_initial",
DataTypes.STRING())),
+ TableChange.modifyColumnPosition(
+ physical("user_count",
DataTypes.BIGINT()),
+
ColumnPosition.after("name_initial")))),
+ TestSpec.of(
+ "non-persisted column between physicals does not skew
the position diff",
+ schema(
+ physical("city", DataTypes.STRING()),
+ metadata(
+ "ingest_time",
+ DataTypes.TIMESTAMP_LTZ(3),
+ "timestamp",
+ true),
+ physical("user_count", DataTypes.BIGINT())),
+ schema(
+ physical("city", DataTypes.STRING()),
+ physical("name_initial", DataTypes.STRING()),
+ physical("user_count", DataTypes.BIGINT())),
+ false,
+ List.of(
+ TableChange.add(physical("name_initial",
DataTypes.STRING())),
+ TableChange.modifyColumnPosition(
+ physical("user_count",
DataTypes.BIGINT()),
+
ColumnPosition.after("name_initial")))),
+ TestSpec.of(
+ "query reorders existing columns, positions emitted",
+ schema(physical("a", DataTypes.INT()), physical("b",
DataTypes.STRING())),
+ schema(physical("b", DataTypes.STRING()),
physical("a", DataTypes.INT())),
+ false,
+ List.of(
+ TableChange.modifyColumnPosition(
+ physical("b", DataTypes.STRING()),
ColumnPosition.first()),
+ TableChange.modifyColumnPosition(
+ physical("a", DataTypes.INT()),
+ ColumnPosition.after("b")))));
}
private static ResolvedSchema schema(Column... columns) {
