raminqaf commented on code in PR #29034:
URL: https://github.com/apache/flink/pull/29034#discussion_r3911601752
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java:
##########
@@ -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);
+ }
Review Comment:
Can we parametize this test and
`testCreateOrAlterMaterializedTableRejectsColumnListReorderOnAlterPath`? They
are very similar to each other and only the `materializedTableDDL` is different
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java:
##########
@@ -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]."),
Review Comment:
A bit confused here why this case was removed. If I understood correctly, we
are replacing a nullable column with a non-nullable column. Did you removed it
because it is a valid case now?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java:
##########
@@ -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.
+ */
Review Comment:
This javadocs is very hard to read some how...
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/MaterializedTableUtils.java:
##########
@@ -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());
Review Comment:
Should we add a proper JavaDocs on this public method explaining in bullet
points what validations we do?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationSqlCreateOrAlterMaterializedTableConverterTest.java:
##########
@@ -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() {
Review Comment:
```suggestion
void
testAlterMaterializedTableAsQueryInsertingComputedColumnBeforeColumns() {
```
--
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]
