From 745323d30107e1072c0603c5e271e43a93de21c9 Mon Sep 17 00:00:00 2001 From: Arvid Heise <4559103+AHeise@users.noreply.github.com> Date: Tue, 1 Sep 2026 17:17:51 +0200 Subject: [PATCH] [FLINK-40496][table-planner] Apply append-only column rules consistently when altering a materialized table 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 rejected consistently; a query-inferred nullability change is treated directionally - a tightening flip such as STRING to STRING NOT NULL is tolerated as an inference artifact while a loosening flip from NOT NULL to nullable renders as a physical column type change and is rejected. Co-Authored-By: Claude Opus 4.8 --- .../MaterializedTableStatementITCase.java | 48 ++++- ...AlterMaterializedTableChangeOperation.java | 13 +- ...edTableAsQueryOperationValidationTest.java | 31 ++++ ...lterMaterializedTableAsQueryConverter.java | 3 +- .../planner/utils/MaterializedTableUtils.java | 174 +++--------------- ...izedTableNodeToOperationConverterTest.java | 7 - ...OrAlterMaterializedTableConverterTest.java | 31 ++++ .../utils/MaterializedTableUtilsTest.java | 108 ----------- .../ValidateAndExtractColumnChangesTest.java | 60 +++++- 9 files changed, 198 insertions(+), 277 deletions(-) delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/MaterializedTableUtilsTest.java 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 ddc0fcd8500984..0c0e51133b2fd2 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 @@ -1083,7 +1083,7 @@ void testCreateOrAlterMaterializedTableColumnListReordersSchemaOnCreatePath() th } @Test - void testCreateOrAlterMaterializedTableColumnListIgnoredOnAlterPath() throws Exception { + void testCreateOrAlterMaterializedTableRejectsColumnListReorderOnAlterPath() throws Exception { createAndVerifyCreateMaterializedTableWithData( "users_shops", List.of(), Map.of(), RefreshMode.FULL); @@ -1092,8 +1092,8 @@ void testCreateOrAlterMaterializedTableColumnListIgnoredOnAlterPath() throws Exc 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. + // A bare column list orders columns on CREATE, so CREATE OR ALTER honors it on the alter + // path too: ordering the existing columns differently is a reorder and is rejected. String materializedTableDDL = "CREATE OR ALTER MATERIALIZED TABLE users_shops (shop_id, user_id, ds, order_cnt)" + " PARTITIONED BY (ds)\n" @@ -1110,12 +1110,12 @@ void testCreateOrAlterMaterializedTableColumnListIgnoredOnAlterPath() throws Exc + " ) 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); + assertThatThrownBy(() -> awaitOperationTermination(service, sessionHandle, handle)) + .hasStackTraceContaining("reordering columns are not supported"); + + // The rejected statement leaves the original schema untouched. + assertThat(getTable(userShopsIdentifier).getResolvedSchema()).isEqualTo(oldSchema); } @Test @@ -1151,6 +1151,38 @@ void testAlterMaterializedTableAsQueryRejectsReorder() throws Exception { .containsExactly("user_id", "shop_id", "ds", "order_cnt"); } + @Test + void testCreateOrAlterMaterializedTableAsQueryRejectsReorder() 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"); + + // CREATE OR ALTER on an existing table takes the alter path; swapping the first two + // projections reorders existing columns and is rejected, matching ALTER ... AS. + String materializedTableDDL = + "CREATE OR 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(materializedTableDDL); + + assertThatThrownBy(() -> awaitOperationTermination(service, sessionHandle, handle)) + .hasStackTraceContaining("reordering columns are not supported"); + + // The rejected statement leaves the original schema untouched. + assertThat(getTable(userShopsIdentifier).getResolvedSchema()).isEqualTo(oldSchema); + } + @Test void testCreateMaterializedTableWithDistribution() { String materializedTableDDL = 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 299c72d6cfea3b..856b11b55dd11a 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 @@ -117,11 +117,20 @@ public void setOldTable(final ResolvedCatalogMaterializedTable oldTable) { this.newTable = null; } + /** + * Whether the query-column rules (no reordering or retyping of existing columns) apply. CREATE + * OR ALTER and ALTER ... AS carry the defining query, so the rules apply even when the query + * text is unchanged; metadata-only DDL alters carry no query. + */ + private boolean appliesQueryColumnRules(List changes) { + return asQueryOperation != null + || changes.stream().anyMatch(ModifyDefinitionQuery.class::isInstance); + } + @VisibleForTesting public void validateChanges() { final List changes = getTableChanges(); - final boolean isQueryChange = - changes.stream().anyMatch(ModifyDefinitionQuery.class::isInstance); + final boolean isQueryChange = appliesQueryColumnRules(changes); final List oldColumns = oldTable.getResolvedSchema().getColumns(); final Map 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 88073daca8bd10..47a0b5c1b6d633 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 @@ void rejectTypeChange() { "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 eb631523b98c3b..e6f922984bc666 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 @@ protected Function> gatherTa ResolvedSchema newSchema = queryOperation.getResolvedSchema(); List 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 69c7277ab1aed4..663799f0de5855 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 static RefreshMode fromLogicalRefreshModeToRefreshMode( } } - // Used to build changes introduced by changed query like - // ALTER MATERIALIZED TABLE ... AS ... - public static List buildSchemaTableChanges( - ResolvedSchema oldSchema, ResolvedSchema newSchema) { - if (!isSchemaChanged(oldSchema, newSchema)) { - return List.of(); - } - - final List oldColumns = oldSchema.getColumns(); - final Map> 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 newColumns = newSchema.getColumns(); - - List changes = new ArrayList<>(); - for (int i = 0; i < newColumns.size(); i++) { - Column newColumn = newColumns.get(i); - Tuple2 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> 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 @@ private static StartModeKind deriveStartModeKind(SqlStartModeKind sqlStartModeKi } } - // 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 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 newColumns, Tuple2 oldColumnToPosition, @@ -417,42 +305,22 @@ private static void applyPositionChanges( } } - private static void applyComputedColumnChanges( - ComputedColumn oldColumn, ComputedColumn newColumn, List 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 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()))); - } - } - public static List validateAndExtractColumnChanges( ResolvedSchema oldSchema, ResolvedSchema newSchema, boolean schemaDefinedInQuery) { final List oldColumns = oldSchema.getColumns(); + final List newColumns = newSchema.getColumns(); + final Set 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> 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 seen = new HashSet<>(); - final List newColumns = newSchema.getColumns(); final List changes = new ArrayList<>(); for (int newIndex = 0; newIndex < newColumns.size(); newIndex++) { final Column newColumn = newColumns.get(newIndex); @@ -463,8 +331,11 @@ public static List validateAndExtractColumnChanges( 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 +412,16 @@ private static boolean typeChanged( 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 a90a58dcd085b2..70f0cf3e106fb7 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 @@ private static Collection alterQuery() { + "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 1eeeeafc627832..82eeea5bbff3c3 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.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 @@ void testAlterMaterializedTableAsQueryWithDefinedSchema() { TableChange.reset("format")); } + /** + * Inserting a new column before more than one trailing column reorders existing columns + * relative to each other. The append-only guard in {@code validateChanges} rejects this - the + * gate that, in OSS, catches what the positional refresh INSERT would otherwise miscompile. + */ + @Test + void testAlterMaterializedTableAsQueryInsertingColumnBeforeMultipleColumns() { + 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 1859c0f531b846..00000000000000 --- 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 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 expected; - - public TestSpec( - ResolvedSchema oldSchema, ResolvedSchema newSchema, List expected) { - - this.oldSchema = oldSchema; - this.newSchema = newSchema; - this.expected = expected; - } - - public static TestSpec of( - ResolvedSchema oldSchema, ResolvedSchema newSchema, List 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 3d3f177af2422d..7ee53c94e91836 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.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 @@ private static Collection input() { 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 @@ private static Collection input() { 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) {