Skip to content

[FLINK-40496][table-planner] Apply append-only column rules consistently when altering a materialized table #29034

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
AHeise wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-40496][table-planner] Apply append-only column rules consistently when altering a materialized table #29034

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 40 additions & 8 deletions ...rc/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand All @@ -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"
Expand All @@ -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
Expand Down Expand Up @@ -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 =
Expand Down
13 changes: 11 additions & 2 deletions ...pache/flink/table/operations/materializedtable/AlterMaterializedTableChangeOperation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TableChange> changes) {
return asQueryOperation != null
|| changes.stream().anyMatch(ModifyDefinitionQuery.class::isInstance);
}

@VisibleForTesting
public void validateChanges() {
final List<TableChange> changes = getTableChanges();
final boolean isQueryChange =
changes.stream().anyMatch(ModifyDefinitionQuery.class::isInstance);
final boolean isQueryChange = appliesQueryColumnRules(changes);
final List<Column> oldColumns = oldTable.getResolvedSchema().getColumns();
final Map<String, Integer> columnIndex =
IntStream.range(0, oldColumns.size())
Expand Down
31 changes: 31 additions & 0 deletions ...le/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down
3 changes: 2 additions & 1 deletion ...er/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,8 @@ protected Function<ResolvedCatalogMaterializedTable, List<TableChange>> gatherTa
ResolvedSchema newSchema = queryOperation.getResolvedSchema();
List<TableChange> tableChanges =
new ArrayList<>(
MaterializedTableUtils.buildSchemaTableChanges(oldSchema, newSchema));
MaterializedTableUtils.validateAndExtractColumnChanges(
oldSchema, newSchema, false));

if (!tableChanges.isEmpty()) {
final boolean hasNonPersistedColumn =
Expand Down
174 changes: 25 additions & 149 deletions ...le-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,91 +262,6 @@ public static RefreshMode fromLogicalRefreshModeToRefreshMode(
}
}

// 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
Expand Down Expand Up @@ -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<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,
Expand All @@ -417,42 +305,22 @@ private static void applyPositionChanges(
}
}

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())));
}
}

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);
Expand All @@ -463,8 +331,11 @@ public static List<TableChange> 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)) {
Expand Down Expand Up @@ -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(
Expand Down
7 changes: 0 additions & 7 deletions ...ache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -849,13 +849,6 @@ private static Collection<TestSpec> 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 "
Expand Down
Loading