snuyanzin commented on code in PR #27302:
URL: https://github.com/apache/flink/pull/27302#discussion_r2588683839
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java:
##########
@@ -160,33 +167,39 @@ public static void validatePhysicalColumnsUsedByQuery(
context.toRelRoot(validateQuery).project(),
() -> context.toQuotedSqlString(validateQuery));
- validatePhysicalColumnsUsedByQuery(sqlNodeList,
queryOperation.getResolvedSchema());
+ validatePersistedColumnsUsedByQuery(sqlNodeList,
queryOperation.getResolvedSchema());
}
- public static void validatePhysicalColumnsUsedByQuery(
+ public static void validatePersistedColumnsUsedByQuery(
SqlNodeList columnPositions, ResolvedSchema querySchema) {
final Set<String> querySchemaColumnNames = new
HashSet<>(querySchema.getColumnNames());
for (SqlNode column : columnPositions) {
- throwIfPhysicalColumnNotUsedByQuery(column,
querySchemaColumnNames);
+ throwIfPersistedColumnNotUsedByQuery(column,
querySchemaColumnNames);
}
}
- private static void throwIfPhysicalColumnNotUsedByQuery(
+ private static void throwIfPersistedColumnNotUsedByQuery(
SqlNode column, Set<String> querySchemaColumnNames) {
- if (column instanceof SqlTableColumn.SqlRegularColumn) {
- SqlTableColumn.SqlRegularColumn physicalColumn =
- (SqlTableColumn.SqlRegularColumn) column;
- if
(!querySchemaColumnNames.contains(physicalColumn.getName().getSimple())) {
- throw new ValidationException(
- String.format(
- "Failed to execute ALTER MATERIALIZED TABLE
statement.\n"
- + "Invalid schema change. All physical
columns in the schema part need to be present in the query part. "
- + "However, column `%s` could not be
found in the query.",
- physicalColumn.getName().getSimple()));
+ if (column instanceof SqlRegularColumn) {
+ SqlRegularColumn physicalColumn = (SqlRegularColumn) column;
+ String columnName = physicalColumn.getName().getSimple();
+ if (!querySchemaColumnNames.contains(columnName)) {
+ throwPersistedColumnNotUsedException("physical", columnName);
+ }
+ } else if (column instanceof SqlMetadataColumn) {
+ SqlMetadataColumn metadataColumn = (SqlMetadataColumn) column;
+ String columnName = metadataColumn.getName().getSimple();
+ if (!metadataColumn.isVirtual() &&
!querySchemaColumnNames.contains(columnName)) {
Review Comment:
There is no `isPersisted` method for `SqlMetadataColumn` however the idea is
same as at
https://github.com/apache/flink/blob/b61de59e37f08818d78afc17cf5ff7b6a5198c05/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/api/TableColumn.java#L291-L294
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
