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) {