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 b806823a648 [FLINK-40267][table-planner] Validate persisted columns on
CREATE MATERIALIZED TABLE
b806823a648 is described below
commit b806823a64863274161c3dfdac1e6ef9a459ff64
Author: Arvid Heise <[email protected]>
AuthorDate: Fri Jul 31 09:13:05 2026 +0200
[FLINK-40267][table-planner] Validate persisted columns on CREATE
MATERIALIZED TABLE
Persisted columns of a materialized table have to be produced by its
definition query, and CREATE OR ALTER enforces that through
validatePersistedColumnsUsedByQuery while a plain CREATE only checked physical
columns. A CREATE declaring a non-virtual metadata column that the query does
not project was therefore accepted, and the mismatch only surfaced when the
refresh pipeline compiled the insert, whose consumed row type contains the
persisted metadata column.
Both statements share the validation now, so the DDL is rejected up front,
and the error message names the statement it originates from. The test fixture
for materialized tables with metadata columns projects its persisted column,
which makes dropping that column report that it is used in the query rather
than that persisted columns cannot be dropped; the latter stays covered by the
query-evolution cases.
Generated-by: Claude Opus 5
---
.../SqlAlterMaterializedTableSchemaConverter.java | 4 +-
...SqlCreateOrAlterMaterializedTableConverter.java | 30 ++------
.../planner/utils/MaterializedTableUtils.java | 35 +++++----
...erializedTableNodeToOperationConverterTest.java | 82 ++++++++++++++++++----
4 files changed, 91 insertions(+), 60 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableSchemaConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableSchemaConverter.java
index 05ffdd3ce45..c13e7d8706b 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableSchemaConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableSchemaConverter.java
@@ -57,6 +57,7 @@ import java.util.function.Function;
public abstract class SqlAlterMaterializedTableSchemaConverter<
T extends SqlAlterMaterializedTableSchema>
extends AbstractAlterMaterializedTableConverter<T> {
+
@Override
protected Operation convertToOperation(
T alterTableSchema, ResolvedCatalogMaterializedTable oldTable,
ConvertContext context) {
@@ -69,9 +70,10 @@ public abstract class
SqlAlterMaterializedTableSchemaConverter<
@Override
protected Function<ResolvedCatalogMaterializedTable, List<TableChange>>
gatherTableChanges(
T alterTableSchema, ConvertContext context) {
+
return oldTable -> {
MaterializedTableUtils.validatePersistedColumnsUsedByQuery(
- oldTable, alterTableSchema, context);
+ oldTable, alterTableSchema, context,
alterTableSchema.getOperator().getName());
SchemaConverter converter = createSchemaConverter(oldTable,
context);
converter.updateColumn(alterTableSchema.getColumnPositions().getList());
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlCreateOrAlterMaterializedTableConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlCreateOrAlterMaterializedTableConverter.java
index bb757cd91a4..cfc7d87d6a3 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlCreateOrAlterMaterializedTableConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlCreateOrAlterMaterializedTableConverter.java
@@ -49,16 +49,13 @@ import
org.apache.flink.table.planner.operations.PlannerQueryOperation;
import org.apache.flink.table.planner.operations.converters.MergeTableAsUtil;
import org.apache.flink.table.planner.utils.MaterializedTableUtils;
-import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlNodeList;
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
-import java.util.Set;
/** A converter for {@link SqlCreateOrAlterMaterializedTable}. */
public class SqlCreateOrAlterMaterializedTableConverter
@@ -443,12 +440,10 @@ public class SqlCreateOrAlterMaterializedTableConverter
@Override
public Schema getMergedSchema() {
final SqlNodeList sqlNodeList =
sqlCreateMaterializedTable.getColumnList();
- if (createOrAlterOperation(sqlCreateMaterializedTable)) {
- MaterializedTableUtils.validatePersistedColumnsUsedByQuery(
- sqlNodeList, querySchema);
- } else {
- validatePhysicalColumnsUsedByQuery(sqlNodeList,
querySchema);
- }
+ MaterializedTableUtils.validatePersistedColumnsUsedByQuery(
+ sqlNodeList,
+ querySchema,
+ sqlCreateMaterializedTable.getOperator().getName());
if
(sqlCreateMaterializedTable.isSchemaWithColumnsIdentifiersOnly()) {
// If only column identifiers are provided, then these are
used to
// order the columns in the schema.
@@ -525,21 +520,4 @@ public class SqlCreateOrAlterMaterializedTableConverter
}
};
}
-
- private static void validatePhysicalColumnsUsedByQuery(
- SqlNodeList sqlNodeList, ResolvedSchema querySchema) {
- final Set<String> querySchemaColumnNames = new
HashSet<>(querySchema.getColumnNames());
- for (SqlNode column : sqlNodeList) {
- if (!(column instanceof SqlRegularColumn)) {
- continue;
- }
- final SqlRegularColumn physicalColumn = (SqlRegularColumn) column;
- if
(!querySchemaColumnNames.contains(physicalColumn.getName().getSimple())) {
- throw new ValidationException(
- String.format(
- "Invalid as physical column '%s' is defined in
the DDL, but is not used in a query column.",
- physicalColumn.getName().getSimple()));
- }
- }
- }
}
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 e454eb7dd8c..69c7277ab1a 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
@@ -73,8 +73,8 @@ import static java.time.temporal.ChronoField.MONTH_OF_YEAR;
public class MaterializedTableUtils {
private static final String PERSISTED_COLUMN_NOT_USED_IN_QUERY =
- "Failed to execute ALTER MATERIALIZED TABLE statement.\n"
- + "Invalid schema change. All persisted (physical and
metadata) columns "
+ "Failed to execute %s statement.\n"
+ + "Invalid schema. All persisted (physical and metadata)
columns "
+ "in the schema part need to be present in the query
part.\n"
+ "However, %s column `%s` could not be found in the
query.";
@@ -563,52 +563,51 @@ public class MaterializedTableUtils {
public static void validatePersistedColumnsUsedByQuery(
CatalogMaterializedTable oldTable,
SqlAlterMaterializedTableSchema alterTableSchema,
- ConvertContext context) {
+ ConvertContext context,
+ String operationName) {
final SqlNodeList sqlNodeList = alterTableSchema.getColumnPositions();
if (sqlNodeList.isEmpty()) {
return;
}
final ResolvedSchema querySchema =
getQueryOperationResolvedSchema(oldTable, context);
- validatePersistedColumnsUsedByQuery(sqlNodeList, querySchema);
+ validatePersistedColumnsUsedByQuery(sqlNodeList, querySchema,
operationName);
}
public static void validatePersistedColumnsUsedByQuery(
- SqlNodeList columnPositions, ResolvedSchema querySchema) {
+ SqlNodeList columnPositions, ResolvedSchema querySchema, String
operationName) {
final Set<String> querySchemaColumnNames = new
HashSet<>(querySchema.getColumnNames());
for (SqlNode column : columnPositions) {
- throwIfPersistedColumnNotUsedByQuery(column,
querySchemaColumnNames);
+ throwIfPersistedColumnNotUsedByQuery(column,
querySchemaColumnNames, operationName);
}
}
private static void throwIfPersistedColumnNotUsedByQuery(
- SqlNode column, Set<String> querySchemaColumnNames) {
+ SqlNode column, Set<String> querySchemaColumnNames, String
operationName) {
if (column instanceof SqlRegularColumn) {
String columnName = ((SqlRegularColumn)
column).getName().getSimple();
if (!querySchemaColumnNames.contains(columnName)) {
- throwPersistedColumnNotUsedException("physical", columnName);
+ throwPersistedColumnNotUsedException(operationName,
"physical", columnName);
}
} else if (column instanceof SqlMetadataColumn) {
SqlMetadataColumn metadataColumn = (SqlMetadataColumn) column;
String columnName = metadataColumn.getName().getSimple();
if (!metadataColumn.isVirtual() &&
!querySchemaColumnNames.contains(columnName)) {
- throwPersistedColumnNotUsedException("metadata persisted",
columnName);
+ throwPersistedColumnNotUsedException(
+ operationName, "metadata persisted", columnName);
}
} else if (column instanceof SqlTableColumnPosition) {
throwIfPersistedColumnNotUsedByQuery(
- ((SqlTableColumnPosition) column).getColumn(),
querySchemaColumnNames);
+ ((SqlTableColumnPosition) column).getColumn(),
+ querySchemaColumnNames,
+ operationName);
}
}
- private static List<Column> getPersistedColumns(ResolvedSchema schema) {
- return schema.getColumns().stream()
- .filter(Column::isPersisted)
- .collect(Collectors.toList());
- }
-
- private static void throwPersistedColumnNotUsedException(String type,
String columnName) {
+ private static void throwPersistedColumnNotUsedException(
+ String operationName, String type, String columnName) {
throw new ValidationException(
- String.format(PERSISTED_COLUMN_NOT_USED_IN_QUERY, type,
columnName));
+ String.format(PERSISTED_COLUMN_NOT_USED_IN_QUERY,
operationName, type, columnName));
}
private static void validateIntervalValuePositive(
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 3c4aeac8343..a90a58dcd08 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
@@ -154,7 +154,8 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ ")\n"
+ "FRESHNESS = INTERVAL '30' SECOND\n"
+ "REFRESH_MODE = FULL\n"
- + "AS SELECT t1.* FROM t1";
+ // the persisted metadata column has to be produced by
the query
+ + "AS SELECT CAST(NULL AS STRING) AS m_p, t1.* FROM
t1";
createMaterializedTableInCatalog(sqlWithMetadataColumn,
"base_mtbl_with_metadata");
// MATERIALIZED TABLE without constraint
@@ -879,8 +880,8 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ "`t` AS CURRENT_TIMESTAMP - INTERVAL '5'
SECOND. You might want to drop it before adding a new one."),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl ADD
`physical_not_used_in_query` BIGINT NOT NULL",
- "Failed to execute ALTER MATERIALIZED TABLE
statement.\n"
- + "Invalid schema change. All persisted
(physical and metadata) "
+ "Failed to execute ALTER MATERIALIZED TABLE ADD
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ "columns in the schema part need to be
present in the query part.\n"
+ "However, physical column
`physical_not_used_in_query` could not be found in the query."),
TestSpec.of(
@@ -902,8 +903,8 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ "Referenced column `q2` by 'AFTER' does not
exist in the table."),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl ADD `m1` INT
METADATA",
- "Failed to execute ALTER MATERIALIZED TABLE
statement.\n"
- + "Invalid schema change. All persisted
(physical and metadata) "
+ "Failed to execute ALTER MATERIALIZED TABLE ADD
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ "columns in the schema part need to be
present in the query part.\n"
+ "However, metadata persisted column `m1`
could not be found in the query."));
}
@@ -920,14 +921,14 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ "Invalid column name 'invalid_column' for
rowtime attribute in watermark declaration. Available columns are: [t, a, b, c,
d]"),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl MODIFY
`physical_not_used_in_query` BIGINT NOT NULL",
- "Failed to execute ALTER MATERIALIZED TABLE
statement.\n"
- + "Invalid schema change. All persisted
(physical and metadata) "
+ "Failed to execute ALTER MATERIALIZED TABLE MODIFY
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ "columns in the schema part need to be
present in the query part.\n"
+ "However, physical column
`physical_not_used_in_query` could not be found in the query."),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl MODIFY
`not_existed_column` BIGINT NOT NULL",
- "Failed to execute ALTER MATERIALIZED TABLE
statement.\n"
- + "Invalid schema change. All persisted
(physical and metadata) columns in the schema part need to be present in the
query part.\n"
+ "Failed to execute ALTER MATERIALIZED TABLE MODIFY
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) columns in the schema part need to be present in the query part.\n"
+ "However, physical column
`not_existed_column` could not be found in the query."),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl MODIFY `a` AS
`non_existing_column` + 2",
@@ -947,8 +948,8 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ "Referenced column `q2` by 'AFTER' does not
exist in the table."),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl MODIFY `m1` INT
METADATA",
- "Failed to execute ALTER MATERIALIZED TABLE
statement.\n"
- + "Invalid schema change. All persisted
(physical and metadata) "
+ "Failed to execute ALTER MATERIALIZED TABLE MODIFY
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ "columns in the schema part need to be
present in the query part.\n"
+ "However, metadata persisted column `m1`
could not be found in the query."),
TestSpec.of(
@@ -1009,12 +1010,42 @@ class SqlMaterializedTableNodeToOperationConverterTest
"CREATE MATERIALIZED TABLE users_shops (a INT, b INT)"
+ " FRESHNESS = INTERVAL '30' SECOND"
+ " AS SELECT 1 AS shop_id, 2 AS user_id",
- "Invalid as physical column 'a' is defined in the DDL,
but is not used in a query column."),
+ "Failed to execute CREATE MATERIALIZED TABLE
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ + "columns in the schema part need to be
present in the query part.\n"
+ + "However, physical column `a` could not be
found in the query."),
TestSpec.of(
"CREATE MATERIALIZED TABLE users_shops (shop_id INT, b
INT)"
+ " FRESHNESS = INTERVAL '30' SECOND"
+ " AS SELECT 1 AS shop_id, 2 AS user_id",
- "Invalid as physical column 'b' is defined in the DDL,
but is not used in a query column."),
+ "Failed to execute CREATE MATERIALIZED TABLE
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ + "columns in the schema part need to be
present in the query part.\n"
+ + "However, physical column `b` could not be
found in the query."),
+ TestSpec.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id INT,
user_id INT, m1 INT METADATA)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Failed to execute CREATE MATERIALIZED TABLE
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ + "columns in the schema part need to be
present in the query part.\n"
+ + "However, metadata persisted column `m1`
could not be found in the query."),
+ TestSpec.of(
+ "CREATE MATERIALIZED TABLE users_shops (shop_id INT,
user_id INT, m1 INT METADATA FROM 'k1')"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Failed to execute CREATE MATERIALIZED TABLE
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ + "columns in the schema part need to be
present in the query part.\n"
+ + "However, metadata persisted column `m1`
could not be found in the query."),
+ TestSpec.of(
+ "CREATE OR ALTER MATERIALIZED TABLE users_shops
(shop_id INT, user_id INT, m1 INT METADATA)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ "Failed to execute CREATE OR ALTER MATERIALIZED TABLE
statement.\n"
+ + "Invalid schema. All persisted (physical and
metadata) "
+ + "columns in the schema part need to be
present in the query part.\n"
+ + "However, metadata persisted column `m1`
could not be found in the query."),
// test unsupported constraint
TestSpec.of(
"CREATE MATERIALIZED TABLE mtbl1 (\n"
@@ -1161,7 +1192,8 @@ class SqlMaterializedTableNodeToOperationConverterTest
+ "Column(s) ('d') are used in query."),
TestSpec.of(
"ALTER MATERIALIZED TABLE base_mtbl_with_metadata DROP
m_p",
- "Dropping of persisted column `m_p` is not
supported."));
+ "Failed to execute ALTER MATERIALIZED TABLE
statement.\n"
+ + "Column(s) ('m_p') are used in query."));
}
private static Collection<TestSpec> alterSet() {
@@ -1475,7 +1507,27 @@ class SqlMaterializedTableNodeToOperationConverterTest
List.of(),
UniqueConstraint.primaryKey("PK_user_id",
List.of("user_id")),
List.of(),
- null)));
+ null)),
+ // a persisted metadata column is written by the query column
of the same name
+ Arguments.of(
+ operation
+ + "MATERIALIZED TABLE users_shops (m1 INT
METADATA)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id, 3 AS
m1",
+ ResolvedSchema.of(
+ Column.physical("shop_id",
DataTypes.INT().notNull()),
+ Column.physical("user_id",
DataTypes.INT().notNull()),
+ Column.metadata("m1", DataTypes.INT(), null,
false))),
+ // a virtual metadata column is read-only and needs no query
column
+ Arguments.of(
+ operation
+ + "MATERIALIZED TABLE users_shops (m2 STRING
METADATA VIRTUAL)"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT 1 AS shop_id, 2 AS user_id",
+ ResolvedSchema.of(
+ Column.metadata("m2", DataTypes.STRING(),
null, true),
+ Column.physical("shop_id",
DataTypes.INT().notNull()),
+ Column.physical("user_id",
DataTypes.INT().notNull()))));
}
/** Boilerplate CatalogMaterializedTable builder for tests. */
