You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2025/01/09 02:43:18 UTC
(flink) branch master updated (329ad1e22b6 -> 5e0e5a233eb)
This is an automated email from the ASF dual-hosted git repository.
ron pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 329ad1e22b6 [FLINK-37029][docs] Fix the materialized table docs of full mode cron expression (#25917)
new 342e3bf1272 [FLINK-36994][table] Support parsing ALTER MATERIALIZED TABLE AS statement
new 9260b33cc51 [FLINK-36994][table] Support converting ALTER MATERIALIZED TABLE AS node to operation
new 44208050673 [FLINK-36994][table] Support executing ALTER MATERIALIZED TABLE AS operation
new 5e0e5a233eb [FLINK-36994][table] Fix expand sql error when materialized table creation with UDTF queries
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../MaterializedTableManager.java | 173 +++++++++-
.../service/MaterializedTableStatementITCase.java | 349 +++++++++++++++++++++
.../src/main/codegen/data/Parser.tdd | 1 +
.../src/main/codegen/includes/parserImpls.ftl | 10 +
....java => SqlAlterMaterializedTableAsQuery.java} | 21 +-
.../MaterializedTableStatementParserTest.java | 22 ++
...=> AlterMaterializedTableAsQueryOperation.java} | 32 +-
.../AlterMaterializedTableChangeOperation.java | 28 +-
.../apache/flink/table/catalog/TableChange.java | 93 +++++-
.../SqlAlterMaterializedTableAsQueryConverter.java | 161 ++++++++++
.../SqlCreateMaterializedTableConverter.java | 14 +-
.../operations/converters/SqlNodeConverters.java | 1 +
...erializedTableNodeToOperationConverterTest.java | 185 ++++++++++-
13 files changed, 1038 insertions(+), 52 deletions(-)
copy flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/{SqlAlterMaterializedTableSuspend.java => SqlAlterMaterializedTableAsQuery.java} (70%)
copy flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/{AlterMaterializedTableSuspendOperation.java => AlterMaterializedTableAsQueryOperation.java} (52%)
create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterMaterializedTableAsQueryConverter.java
(flink) 02/04: [FLINK-36994][table] Support converting ALTER MATERIALIZED TABLE AS node to operation
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9260b33cc5109c6dd2042a8f0ce645b6d56d4d57
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Sun Jan 5 20:05:38 2025 +0800
[FLINK-36994][table] Support converting ALTER MATERIALIZED TABLE AS node to operation
---
.../AlterMaterializedTableAsQueryOperation.java | 66 +++++++++
.../AlterMaterializedTableChangeOperation.java | 28 ++--
.../apache/flink/table/catalog/TableChange.java | 93 ++++++++++--
.../SqlAlterMaterializedTableAsQueryConverter.java | 161 +++++++++++++++++++++
.../operations/converters/SqlNodeConverters.java | 1 +
...erializedTableNodeToOperationConverterTest.java | 143 ++++++++++++++++++
6 files changed, 461 insertions(+), 31 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java
new file mode 100644
index 00000000000..331e8cd0220
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java
@@ -0,0 +1,66 @@
+/*
+ * 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.operations.materializedtable;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.TableChange.MaterializedTableChange;
+
+import java.util.List;
+
+/** Operation to describe an ALTER MATERIALIZED TABLE AS query operation. */
+@Internal
+public class AlterMaterializedTableAsQueryOperation extends AlterMaterializedTableOperation {
+
+ private final List<MaterializedTableChange> tableChanges;
+
+ private final CatalogMaterializedTable newMaterializedTable;
+
+ public AlterMaterializedTableAsQueryOperation(
+ ObjectIdentifier tableIdentifier,
+ List<MaterializedTableChange> tableChanges,
+ CatalogMaterializedTable newMaterializedTable) {
+ super(tableIdentifier);
+ this.tableChanges = tableChanges;
+ this.newMaterializedTable = newMaterializedTable;
+ }
+
+ public List<MaterializedTableChange> getTableChanges() {
+ return tableChanges;
+ }
+
+ public CatalogMaterializedTable getNewMaterializedTable() {
+ return newMaterializedTable;
+ }
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ throw new UnsupportedOperationException(
+ "AlterMaterializedTableAsQueryOperation doesn't support ExecutableOperation yet.");
+ }
+
+ @Override
+ public String asSummaryString() {
+ return String.format(
+ "ALTER MATERIALIZED TABLE %s AS %s",
+ tableIdentifier.asSummaryString(), newMaterializedTable.getDefinitionQuery());
+ }
+}
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 49f220a8ddc..d71946c7a56 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
@@ -24,6 +24,7 @@ import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableChange.MaterializedTableChange;
import org.apache.flink.table.operations.ddl.AlterTableChangeOperation;
import java.util.List;
@@ -35,19 +36,19 @@ import java.util.stream.Collectors;
@Internal
public class AlterMaterializedTableChangeOperation extends AlterMaterializedTableOperation {
- private final List<TableChange> tableChanges;
+ private final List<MaterializedTableChange> tableChanges;
private final CatalogMaterializedTable catalogMaterializedTable;
public AlterMaterializedTableChangeOperation(
ObjectIdentifier tableIdentifier,
- List<TableChange> tableChanges,
+ List<MaterializedTableChange> tableChanges,
CatalogMaterializedTable catalogMaterializedTable) {
super(tableIdentifier);
this.tableChanges = tableChanges;
this.catalogMaterializedTable = catalogMaterializedTable;
}
- public List<TableChange> getTableChanges() {
+ public List<MaterializedTableChange> getTableChanges() {
return tableChanges;
}
@@ -72,22 +73,13 @@ public class AlterMaterializedTableChangeOperation extends AlterMaterializedTabl
public String asSummaryString() {
String changes =
tableChanges.stream()
- .map(
- tableChange -> {
- if (tableChange
- instanceof TableChange.MaterializedTableChange) {
- return toString(
- (TableChange.MaterializedTableChange) tableChange);
- } else {
- return AlterTableChangeOperation.toString(tableChange);
- }
- })
+ .map(AlterMaterializedTableChangeOperation::toString)
.collect(Collectors.joining(",\n"));
return String.format(
"ALTER MATERIALIZED TABLE %s\n%s", tableIdentifier.asSummaryString(), changes);
}
- private String toString(TableChange.MaterializedTableChange tableChange) {
+ private static String toString(MaterializedTableChange tableChange) {
if (tableChange instanceof TableChange.ModifyRefreshStatus) {
TableChange.ModifyRefreshStatus refreshStatus =
(TableChange.ModifyRefreshStatus) tableChange;
@@ -99,9 +91,13 @@ public class AlterMaterializedTableChangeOperation extends AlterMaterializedTabl
return String.format(
" MODIFY REFRESH HANDLER DESCRIPTION TO '%s'",
refreshHandler.getRefreshHandlerDesc());
+ } else if (tableChange instanceof TableChange.ModifyDefinitionQuery) {
+ TableChange.ModifyDefinitionQuery definitionQuery =
+ (TableChange.ModifyDefinitionQuery) tableChange;
+ return String.format(
+ " MODIFY DEFINITION QUERY TO '%s'", definitionQuery.getDefinitionQuery());
} else {
- throw new UnsupportedOperationException(
- String.format("Unknown materialized table change: %s.", tableChange));
+ return AlterTableChangeOperation.toString(tableChange);
}
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableChange.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableChange.java
index f5531f3bcf3..76ac98bd38d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableChange.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableChange.java
@@ -28,7 +28,7 @@ import javax.annotation.Nullable;
import java.util.Arrays;
import java.util.Objects;
-/** {@link TableChange} represents the modification of the table. */
+/** {@link TableChange} represents the modification of the {@link CatalogBaseTable}. */
@PublicEvolving
public interface TableChange {
@@ -380,6 +380,16 @@ public interface TableChange {
return new ModifyRefreshHandler(refreshHandlerDesc, refreshHandlerBytes);
}
+ /**
+ * A table change to modify materialized table definition query.
+ *
+ * @param definitionQuery the modified definition query.
+ * @return a TableChange represents the modification.
+ */
+ static ModifyDefinitionQuery modifyDefinitionQuery(String definitionQuery) {
+ return new ModifyDefinitionQuery(definitionQuery);
+ }
+
// --------------------------------------------------------------------------------------------
// Add Change
// --------------------------------------------------------------------------------------------
@@ -392,9 +402,13 @@ public interface TableChange {
* <pre>
* ALTER TABLE <table_name> ADD <column_definition> <column_position>
* </pre>
+ *
+ * <p>Note: An <code>ALTER MATERIALIZED TABLE AS QUERY</code> operation may also produce an
+ * <code>AddColumn</code> change. This occurs when the materialized table's schema is updated to
+ * align with the structure of the query results, which might require adding new columns.
*/
@PublicEvolving
- class AddColumn implements TableChange {
+ class AddColumn implements CatalogTableChange, MaterializedTableChange {
private final Column column;
private final ColumnPosition position;
@@ -447,7 +461,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class AddUniqueConstraint implements TableChange {
+ class AddUniqueConstraint implements CatalogTableChange {
private final UniqueConstraint constraint;
@@ -493,7 +507,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class AddDistribution implements TableChange {
+ class AddDistribution implements CatalogTableChange {
private final TableDistribution distribution;
@@ -539,7 +553,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class AddWatermark implements TableChange {
+ class AddWatermark implements CatalogTableChange {
private final WatermarkSpec watermarkSpec;
@@ -601,7 +615,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class ModifyColumn implements TableChange {
+ class ModifyColumn implements CatalogTableChange {
protected final Column oldColumn;
protected final Column newColumn;
@@ -848,7 +862,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class ModifyUniqueConstraint implements TableChange {
+ class ModifyUniqueConstraint implements CatalogTableChange {
private final UniqueConstraint newConstraint;
@@ -894,7 +908,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class ModifyDistribution implements TableChange {
+ class ModifyDistribution implements CatalogTableChange {
private final TableDistribution distribution;
@@ -940,7 +954,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class ModifyWatermark implements TableChange {
+ class ModifyWatermark implements CatalogTableChange {
private final WatermarkSpec newWatermark;
@@ -988,9 +1002,14 @@ public interface TableChange {
* <pre>
* ALTER TABLE <table_name> DROP COLUMN <column_name>
* </pre>
+ *
+ * <p>Note: A <code>DropColumn</code> change may also occur when rolling back the schema during
+ * a failed <code>ALTER MATERIALIZED TABLE AS QUERY</code> operation. If the operation fails,
+ * columns added to align with the query results may need to be removed to restore the original
+ * schema.
*/
@PublicEvolving
- class DropColumn implements TableChange {
+ class DropColumn implements CatalogTableChange, MaterializedTableChange {
private final String columnName;
@@ -1036,7 +1055,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class DropWatermark implements TableChange {
+ class DropWatermark implements CatalogTableChange {
static final DropWatermark INSTANCE = new DropWatermark();
@Override
@@ -1055,7 +1074,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class DropConstraint implements TableChange {
+ class DropConstraint implements CatalogTableChange {
private final String constraintName;
@@ -1101,7 +1120,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class DropDistribution implements TableChange {
+ class DropDistribution implements CatalogTableChange {
static final DropDistribution INSTANCE = new DropDistribution();
@Override
@@ -1124,7 +1143,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class SetOption implements TableChange {
+ class SetOption implements CatalogTableChange {
private final String key;
private final String value;
@@ -1177,7 +1196,7 @@ public interface TableChange {
* </pre>
*/
@PublicEvolving
- class ResetOption implements TableChange {
+ class ResetOption implements CatalogTableChange {
private final String key;
@@ -1279,6 +1298,13 @@ public interface TableChange {
}
}
+ // --------------------------------------------------------------------------------------------
+ // Catalog table change
+ // --------------------------------------------------------------------------------------------
+ /** {@link CatalogTableChange} represents the modification of the CatalogTable. */
+ @PublicEvolving
+ interface CatalogTableChange extends TableChange {}
+
// --------------------------------------------------------------------------------------------
// Materialized table change
// --------------------------------------------------------------------------------------------
@@ -1374,4 +1400,41 @@ public interface TableChange {
+ '}';
}
}
+
+ /** A table change to modify the definition query. */
+ @PublicEvolving
+ class ModifyDefinitionQuery implements MaterializedTableChange {
+
+ private final String definitionQuery;
+
+ public ModifyDefinitionQuery(String definitionQuery) {
+ this.definitionQuery = definitionQuery;
+ }
+
+ public String getDefinitionQuery() {
+ return definitionQuery;
+ }
+
+ @Override
+ public String toString() {
+ return "ModifyDefinitionQuery{" + "definitionQuery='" + definitionQuery + '\'' + '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ModifyDefinitionQuery that = (ModifyDefinitionQuery) o;
+ return Objects.equals(definitionQuery, that.definitionQuery);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(definitionQuery);
+ }
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterMaterializedTableAsQueryConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterMaterializedTableAsQueryConverter.java
new file mode 100644
index 00000000000..048e2e4d753
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterMaterializedTableAsQueryConverter.java
@@ -0,0 +1,161 @@
+/*
+ * 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.operations.converters;
+
+import org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTableAsQuery;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableChange.MaterializedTableChange;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableAsQueryOperation;
+import org.apache.flink.table.planner.operations.PlannerQueryOperation;
+
+import org.apache.calcite.sql.SqlNode;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.catalog.CatalogBaseTable.TableKind.MATERIALIZED_TABLE;
+
+/** A converter for {@link SqlAlterMaterializedTableAsQuery}. */
+public class SqlAlterMaterializedTableAsQueryConverter
+ implements SqlNodeConverter<SqlAlterMaterializedTableAsQuery> {
+
+ @Override
+ public Operation convertSqlNode(
+ SqlAlterMaterializedTableAsQuery sqlAlterMaterializedTableAsQuery,
+ ConvertContext context) {
+ ObjectIdentifier identifier = resolveIdentifier(sqlAlterMaterializedTableAsQuery, context);
+
+ // Validate and extract schema from query
+ String originalQuery =
+ context.toQuotedSqlString(sqlAlterMaterializedTableAsQuery.getAsQuery());
+ SqlNode validatedQuery =
+ context.getSqlValidator().validate(sqlAlterMaterializedTableAsQuery.getAsQuery());
+ // The LATERAL operator was eliminated during sql validation, thus the unparsed SQL
+ // does not contain LATERAL which is problematic,
+ // the issue was resolved in CALCITE-4077
+ // (always treat the table function as implicitly LATERAL).
+ String definitionQuery = context.expandSqlIdentifiers(originalQuery);
+ PlannerQueryOperation queryOperation =
+ new PlannerQueryOperation(
+ context.toRelRoot(validatedQuery).project(), () -> originalQuery);
+
+ ResolvedCatalogMaterializedTable oldTable =
+ getResolvedMaterializedTable(context, identifier);
+ List<Column> addedColumns =
+ validateAndExtractNewColumns(
+ oldTable.getResolvedSchema(), queryOperation.getResolvedSchema());
+
+ // Build new materialized table and apply changes
+ CatalogMaterializedTable updatedTable =
+ buildUpdatedMaterializedTable(oldTable, addedColumns, definitionQuery);
+ List<MaterializedTableChange> tableChanges = new ArrayList<>();
+ addedColumns.forEach(column -> tableChanges.add(TableChange.add(column)));
+ tableChanges.add(TableChange.modifyDefinitionQuery(definitionQuery));
+
+ return new AlterMaterializedTableAsQueryOperation(identifier, tableChanges, updatedTable);
+ }
+
+ private ObjectIdentifier resolveIdentifier(
+ SqlAlterMaterializedTableAsQuery sqlAlterTableAsQuery, ConvertContext context) {
+ UnresolvedIdentifier unresolvedIdentifier =
+ UnresolvedIdentifier.of(sqlAlterTableAsQuery.fullTableName());
+ return context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier);
+ }
+
+ private ResolvedCatalogMaterializedTable getResolvedMaterializedTable(
+ ConvertContext context, ObjectIdentifier identifier) {
+ ResolvedCatalogBaseTable<?> baseTable =
+ context.getCatalogManager().getTableOrError(identifier).getResolvedTable();
+ if (MATERIALIZED_TABLE != baseTable.getTableKind()) {
+ throw new ValidationException(
+ "Only materialized table support modify definition query.");
+ }
+ return (ResolvedCatalogMaterializedTable) baseTable;
+ }
+
+ private CatalogMaterializedTable buildUpdatedMaterializedTable(
+ ResolvedCatalogMaterializedTable oldTable,
+ List<Column> addedColumns,
+ String definitionQuery) {
+
+ Schema.Builder newSchemaBuilder =
+ Schema.newBuilder().fromResolvedSchema(oldTable.getResolvedSchema());
+ addedColumns.forEach(col -> newSchemaBuilder.column(col.getName(), col.getDataType()));
+
+ return CatalogMaterializedTable.newBuilder()
+ .schema(newSchemaBuilder.build())
+ .comment(oldTable.getComment())
+ .partitionKeys(oldTable.getPartitionKeys())
+ .options(oldTable.getOptions())
+ .definitionQuery(definitionQuery)
+ .freshness(oldTable.getDefinitionFreshness())
+ .logicalRefreshMode(oldTable.getLogicalRefreshMode())
+ .refreshMode(oldTable.getRefreshMode())
+ .refreshStatus(oldTable.getRefreshStatus())
+ .refreshHandlerDescription(oldTable.getRefreshHandlerDescription().orElse(null))
+ .serializedRefreshHandler(oldTable.getSerializedRefreshHandler())
+ .build();
+ }
+
+ private List<Column> validateAndExtractNewColumns(
+ ResolvedSchema oldSchema, ResolvedSchema newSchema) {
+ List<Column> newAddedColumns = new ArrayList<>();
+ int originalColumnSize = oldSchema.getColumns().size();
+ int newColumnSize = newSchema.getColumns().size();
+
+ if (originalColumnSize > newColumnSize) {
+ throw new ValidationException(
+ String.format(
+ "Failed to modify query because drop column is unsupported. "
+ + "When modifying a query, you can only append new columns at the end of original schema. "
+ + "The original schema has %d columns, but the newly derived schema from the query has %d columns.",
+ originalColumnSize, newColumnSize));
+ }
+
+ for (int i = 0; i < oldSchema.getColumns().size(); i++) {
+ Column oldColumn = oldSchema.getColumns().get(i);
+ Column newColumn = newSchema.getColumns().get(i);
+ if (!oldColumn.equals(newColumn)) {
+ throw new ValidationException(
+ String.format(
+ "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 %d: Original column is [%s], but new column is [%s].",
+ i, oldColumn, newColumn));
+ }
+ }
+
+ for (int i = oldSchema.getColumns().size(); i < newSchema.getColumns().size(); i++) {
+ Column newColumn = newSchema.getColumns().get(i);
+ newAddedColumns.add(newColumn.copy(newColumn.getDataType().nullable()));
+ }
+
+ return newAddedColumns;
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
index 25ab375469f..87570ee9c3c 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
@@ -63,6 +63,7 @@ public class SqlNodeConverters {
register(new SqlAlterMaterializedTableRefreshConverter());
register(new SqlAlterMaterializedTableSuspendConverter());
register(new SqlAlterMaterializedTableResumeConverter());
+ register(new SqlAlterMaterializedTableAsQueryConverter());
register(new SqlDropMaterializedTableConverter());
register(new SqlShowTablesConverter());
register(new SqlShowViewsConverter());
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 ff3cf0c0849..a3b659f335b 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
@@ -28,9 +28,12 @@ import org.apache.flink.table.catalog.IntervalFreshness;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableAsQueryOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableRefreshOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableResumeOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableSuspendOperation;
@@ -45,7 +48,9 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -72,6 +77,25 @@ public class SqlMaterializedTableNodeToOperationConverterTest
final CatalogTable catalogTable =
CatalogTable.of(tableSchema, "", Arrays.asList("b", "c"), options);
catalog.createTable(path3, catalogTable, true);
+
+ // create materialized table
+ final String sql =
+ "CREATE MATERIALIZED TABLE base_mtbl (\n"
+ + " CONSTRAINT ct1 PRIMARY KEY(a) NOT ENFORCED"
+ + ")\n"
+ + "COMMENT 'materialized table comment'\n"
+ + "PARTITIONED BY (a, d)\n"
+ + "WITH (\n"
+ + " 'connector' = 'filesystem', \n"
+ + " 'format' = 'json'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '30' SECOND\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS SELECT * FROM t1";
+ final ObjectPath path4 = new ObjectPath(catalogManager.getCurrentDatabase(), "base_mtbl");
+
+ CreateMaterializedTableOperation operation = (CreateMaterializedTableOperation) parse(sql);
+ catalog.createTable(path4, operation.getCatalogMaterializedTable(), true);
}
@Test
@@ -391,6 +415,125 @@ public class SqlMaterializedTableNodeToOperationConverterTest
.isEqualTo("ALTER MATERIALIZED TABLE builtin.default.mtbl1 RESUME WITH (k1: [v1])");
}
+ @Test
+ void testAlterMaterializedTableAsQuery() throws TableNotExistException {
+ String sql =
+ "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, d, d as e, cast('123' as string) as f FROM t3";
+ Operation operation = parse(sql);
+
+ assertThat(operation).isInstanceOf(AlterMaterializedTableAsQueryOperation.class);
+
+ AlterMaterializedTableAsQueryOperation op =
+ (AlterMaterializedTableAsQueryOperation) operation;
+ assertThat(op.getTableChanges())
+ .isEqualTo(
+ Arrays.asList(
+ TableChange.add(
+ Column.physical("e", DataTypes.VARCHAR(Integer.MAX_VALUE))),
+ TableChange.add(
+ Column.physical("f", DataTypes.VARCHAR(Integer.MAX_VALUE))),
+ TableChange.modifyDefinitionQuery(
+ "SELECT `t3`.`a`, `t3`.`b`, `t3`.`c`, `t3`.`d`, `t3`.`d` AS `e`, CAST('123' AS STRING) AS `f`\n"
+ + "FROM `builtin`.`default`.`t3`")));
+ assertThat(operation.asSummaryString())
+ .isEqualTo(
+ "ALTER MATERIALIZED TABLE builtin.default.base_mtbl AS SELECT `t3`.`a`, `t3`.`b`, `t3`.`c`, `t3`.`d`, `t3`.`d` AS `e`, CAST('123' AS STRING) AS `f`\n"
+ + "FROM `builtin`.`default`.`t3`");
+
+ // new table only difference schema & definition query with old table.
+ CatalogMaterializedTable oldTable =
+ (CatalogMaterializedTable)
+ catalog.getTable(
+ new ObjectPath(catalogManager.getCurrentDatabase(), "base_mtbl"));
+ CatalogMaterializedTable newTable = op.getNewMaterializedTable();
+
+ assertThat(oldTable.getUnresolvedSchema()).isNotEqualTo(newTable.getUnresolvedSchema());
+ assertThat(oldTable.getUnresolvedSchema().getPrimaryKey())
+ .isEqualTo(newTable.getUnresolvedSchema().getPrimaryKey());
+ assertThat(oldTable.getUnresolvedSchema().getWatermarkSpecs())
+ .isEqualTo(newTable.getUnresolvedSchema().getWatermarkSpecs());
+ assertThat(oldTable.getDefinitionQuery()).isNotEqualTo(newTable.getDefinitionQuery());
+ assertThat(oldTable.getDefinitionFreshness()).isEqualTo(newTable.getDefinitionFreshness());
+ assertThat(oldTable.getRefreshMode()).isEqualTo(newTable.getRefreshMode());
+ assertThat(oldTable.getRefreshStatus()).isEqualTo(newTable.getRefreshStatus());
+ assertThat(oldTable.getSerializedRefreshHandler())
+ .isEqualTo(newTable.getSerializedRefreshHandler());
+
+ List<Schema.UnresolvedColumn> addedColumn =
+ newTable.getUnresolvedSchema().getColumns().stream()
+ .filter(
+ column ->
+ !oldTable.getUnresolvedSchema()
+ .getColumns()
+ .contains(column))
+ .collect(Collectors.toList());
+ // added column should be a nullable column.
+ assertThat(addedColumn)
+ .isEqualTo(
+ Arrays.asList(
+ new Schema.UnresolvedPhysicalColumn(
+ "e", DataTypes.VARCHAR(Integer.MAX_VALUE)),
+ new Schema.UnresolvedPhysicalColumn(
+ "f", DataTypes.VARCHAR(Integer.MAX_VALUE))));
+ }
+
+ @Test
+ void testAlterMaterializedTableAsQueryWithConflictColumnName() {
+ String sql5 = "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, d, c as a FROM t3";
+ AlterMaterializedTableAsQueryOperation sqlAlterMaterializedTableAsQuery =
+ (AlterMaterializedTableAsQueryOperation) parse(sql5);
+
+ assertThat(sqlAlterMaterializedTableAsQuery.getTableChanges())
+ .isEqualTo(
+ Arrays.asList(
+ TableChange.add(Column.physical("a0", DataTypes.INT())),
+ TableChange.modifyDefinitionQuery(
+ "SELECT `t3`.`a`, `t3`.`b`, `t3`.`c`, `t3`.`d`, `t3`.`c` AS `a`\n"
+ + "FROM `builtin`.`default`.`t3`")));
+ }
+
+ @Test
+ void testAlterMaterializedTableAsQueryWithUnsupportedColumnChange() {
+ // 1. delete existing column
+ String sql1 = "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b FROM t3";
+ assertThatThrownBy(() -> parse(sql1))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Failed to modify query because drop column is unsupported. When modifying a query, you can only append new columns at the end of original schema. The original schema has 4 columns, but the newly derived schema from the query has 2 columns.");
+ // 2. swap column position
+ String sql2 = "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, d, c FROM t3";
+ assertThatThrownBy(() -> parse(sql2))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "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 2: Original column is [`c` INT], but new column is [`d` STRING].");
+ // 3. change existing column type
+ String sql3 =
+ "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, cast(d as int) as d FROM t3";
+ assertThatThrownBy(() -> parse(sql3))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "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 3: Original column is [`d` STRING], but new column is [`d` INT].");
+ // 4. change existing column nullability
+ String sql4 =
+ "ALTER MATERIALIZED TABLE base_mtbl AS SELECT a, b, c, cast('d' as string) as d FROM t3";
+ assertThatThrownBy(() -> parse(sql4))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "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 3: Original column is [`d` STRING], but new column is [`d` STRING NOT NULL].");
+ }
+
+ @Test
+ void testAlterAlterMaterializedTableAsQueryWithCatalogTable() {
+ // t1 is a CatalogTable not a Materialized Table
+ final String sql = "ALTER MATERIALIZED TABLE t1 AS SELECT * FROM t1";
+ assertThatThrownBy(() -> parse(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Only materialized table support modify definition query.");
+ }
+
@Test
void testDropMaterializedTable() {
final String sql = "DROP MATERIALIZED TABLE mtbl1";
(flink) 01/04: [FLINK-36994][table] Support parsing ALTER MATERIALIZED TABLE AS statement
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 342e3bf1272a5fe448d0bec1405c432658dbf8b4
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Sun Jan 5 20:02:58 2025 +0800
[FLINK-36994][table] Support parsing ALTER MATERIALIZED TABLE AS statement
---
.../src/main/codegen/data/Parser.tdd | 1 +
.../src/main/codegen/includes/parserImpls.ftl | 10 ++++
.../ddl/SqlAlterMaterializedTableAsQuery.java | 58 ++++++++++++++++++++++
.../MaterializedTableStatementParserTest.java | 22 ++++++++
4 files changed, 91 insertions(+)
diff --git a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd
index ff10d47109a..02ba1a1aaaa 100644
--- a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd
+++ b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd
@@ -40,6 +40,7 @@
"org.apache.flink.sql.parser.ddl.SqlAlterDatabase"
"org.apache.flink.sql.parser.ddl.SqlAlterFunction"
"org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTable"
+ "org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTableAsQuery"
"org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTableFreshness"
"org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTableOptions"
"org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTableRefreshMode"
diff --git a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
index b02be2ad72c..f371e6b0dd6 100644
--- a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
+++ b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
@@ -2009,6 +2009,7 @@ SqlAlterMaterializedTable SqlAlterMaterializedTable() :
SqlNodeList propertyKeyList = SqlNodeList.EMPTY;
SqlNodeList partSpec = SqlNodeList.EMPTY;
SqlNode freshness = null;
+ SqlNode asQuery = null;
}
{
<ALTER> <MATERIALIZED> <TABLE> { startPos = getPos();}
@@ -2089,6 +2090,15 @@ SqlAlterMaterializedTable SqlAlterMaterializedTable() :
tableIdentifier,
propertyKeyList);
}
+ |
+ <AS>
+ asQuery = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
+ {
+ return new SqlAlterMaterializedTableAsQuery(
+ startPos.plus(getPos()),
+ tableIdentifier,
+ asQuery);
+ }
)
}
diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterMaterializedTableAsQuery.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterMaterializedTableAsQuery.java
new file mode 100644
index 00000000000..7bc10e2da1a
--- /dev/null
+++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterMaterializedTableAsQuery.java
@@ -0,0 +1,58 @@
+/*
+ * 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.sql.parser.ddl;
+
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.ImmutableNullableList;
+
+import java.util.List;
+
+/**
+ * SqlNode to describe the ALTER TABLE [catalogName.] [dataBasesName.]tableName AS <query>
+ * statement.
+ */
+public class SqlAlterMaterializedTableAsQuery extends SqlAlterMaterializedTable {
+
+ private final SqlNode asQuery;
+
+ public SqlAlterMaterializedTableAsQuery(
+ SqlParserPos pos, SqlIdentifier tableName, SqlNode asQuery) {
+ super(pos, tableName);
+ this.asQuery = asQuery;
+ }
+
+ public SqlNode getAsQuery() {
+ return asQuery;
+ }
+
+ @Override
+ public List<SqlNode> getOperandList() {
+ return ImmutableNullableList.of(getTableName(), asQuery);
+ }
+
+ @Override
+ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+ super.unparse(writer, leftPrec, rightPrec);
+ writer.keyword("AS");
+ asQuery.unparse(writer, leftPrec, rightPrec);
+ }
+}
diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java
index e738060d882..dc4b429206e 100644
--- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java
+++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java
@@ -218,6 +218,7 @@ public class MaterializedTableStatementParserTest {
.fails(
"Encountered \"<EOF>\" at line 1, column 28.\n"
+ "Was expecting one of:\n"
+ + " \"AS\" ...\n"
+ " \"RESET\" ...\n"
+ " \"SET\" ...\n"
+ " \"SUSPEND\" ...\n"
@@ -384,6 +385,27 @@ public class MaterializedTableStatementParserTest {
+ " ");
}
+ @Test
+ void testAlterMaterializedTableAsQuery() {
+ final String sql = "ALTER MATERIALIZED TABLE tbl1 AS SELECT * FROM t";
+ final String expected = "ALTER MATERIALIZED TABLE `TBL1` AS SELECT *\nFROM `T`";
+ sql(sql).ok(expected);
+
+ final String sql2 = "ALTER MATERIALIZED TABLE tbl1 AS SELECT * FROM t A^S^";
+ sql(sql2)
+ .fails(
+ "Encountered \"<EOF>\" at line 1, column 51.\n"
+ + "Was expecting one of:\n"
+ + " <BRACKET_QUOTED_IDENTIFIER> ...\n"
+ + " <QUOTED_IDENTIFIER> ...\n"
+ + " <BACK_QUOTED_IDENTIFIER> ...\n"
+ + " <BIG_QUERY_BACK_QUOTED_IDENTIFIER> ...\n"
+ + " <HYPHENATED_IDENTIFIER> ...\n"
+ + " <IDENTIFIER> ...\n"
+ + " <UNICODE_QUOTED_IDENTIFIER> ...\n"
+ + " ");
+ }
+
@Test
void testDropMaterializedTable() {
final String sql = "DROP MATERIALIZED TABLE tbl1";
(flink) 03/04: [FLINK-36994][table] Support executing ALTER MATERIALIZED TABLE AS operation
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4420805067303841524f7b9987b33c8266b8cb87
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Sun Jan 5 20:06:59 2025 +0800
[FLINK-36994][table] Support executing ALTER MATERIALIZED TABLE AS operation
---
.../MaterializedTableManager.java | 173 +++++++++-
.../service/MaterializedTableStatementITCase.java | 349 +++++++++++++++++++++
2 files changed, 518 insertions(+), 4 deletions(-)
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java
index 2398e72172a..2c2ea78f100 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java
@@ -34,6 +34,7 @@ import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableChange.MaterializedTableChange;
import org.apache.flink.table.data.GenericMapData;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
@@ -48,6 +49,7 @@ import org.apache.flink.table.gateway.service.result.ResultFetcher;
import org.apache.flink.table.gateway.service.utils.SqlExecutionException;
import org.apache.flink.table.operations.command.DescribeJobOperation;
import org.apache.flink.table.operations.command.StopJobOperation;
+import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableAsQueryOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableChangeOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableRefreshOperation;
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableResumeOperation;
@@ -174,6 +176,9 @@ public class MaterializedTableManager {
} else if (op instanceof DropMaterializedTableOperation) {
return callDropMaterializedTableOperation(
operationExecutor, handle, (DropMaterializedTableOperation) op);
+ } else if (op instanceof AlterMaterializedTableAsQueryOperation) {
+ return callAlterMaterializedTableAsQueryOperation(
+ operationExecutor, handle, (AlterMaterializedTableAsQueryOperation) op);
}
throw new SqlExecutionException(
@@ -315,7 +320,7 @@ public class MaterializedTableManager {
return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
}
- private void suspendContinuousRefreshJob(
+ private CatalogMaterializedTable suspendContinuousRefreshJob(
OperationExecutor operationExecutor,
OperationHandle handle,
ObjectIdentifier tableIdentifier,
@@ -341,7 +346,7 @@ public class MaterializedTableManager {
refreshHandler.getJobId(),
savepointPath);
- updateRefreshHandler(
+ return updateRefreshHandler(
operationExecutor,
handle,
tableIdentifier,
@@ -804,6 +809,164 @@ public class MaterializedTableManager {
return insertStatement.toString();
}
+ private ResultFetcher callAlterMaterializedTableAsQueryOperation(
+ OperationExecutor operationExecutor,
+ OperationHandle handle,
+ AlterMaterializedTableAsQueryOperation op) {
+ ObjectIdentifier tableIdentifier = op.getTableIdentifier();
+ CatalogMaterializedTable oldMaterializedTable =
+ getCatalogMaterializedTable(operationExecutor, tableIdentifier);
+
+ if (CatalogMaterializedTable.RefreshMode.FULL == oldMaterializedTable.getRefreshMode()) {
+ // directly apply the alter operation
+ AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
+ new AlterMaterializedTableChangeOperation(
+ tableIdentifier, op.getTableChanges(), op.getNewMaterializedTable());
+ return operationExecutor.callExecutableOperation(
+ handle, alterMaterializedTableChangeOperation);
+ }
+
+ if (CatalogMaterializedTable.RefreshStatus.ACTIVATED
+ == oldMaterializedTable.getRefreshStatus()) {
+ // 1. suspend the materialized table
+ CatalogMaterializedTable suspendMaterializedTable =
+ suspendContinuousRefreshJob(
+ operationExecutor, handle, tableIdentifier, oldMaterializedTable);
+
+ // 2. alter materialized table schema & query definition
+ CatalogMaterializedTable updatedMaterializedTable =
+ op.getNewMaterializedTable()
+ .copy(
+ suspendMaterializedTable.getRefreshStatus(),
+ suspendMaterializedTable
+ .getRefreshHandlerDescription()
+ .orElse(null),
+ suspendMaterializedTable.getSerializedRefreshHandler());
+ AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
+ new AlterMaterializedTableChangeOperation(
+ tableIdentifier, op.getTableChanges(), updatedMaterializedTable);
+ operationExecutor.callExecutableOperation(
+ handle, alterMaterializedTableChangeOperation);
+
+ // 3. resume the materialized table
+ try {
+ executeContinuousRefreshJob(
+ operationExecutor,
+ handle,
+ updatedMaterializedTable,
+ tableIdentifier,
+ Collections.emptyMap(),
+ Optional.empty());
+ } catch (Exception e) {
+ // Roll back the changes to the materialized table and restore the continuous
+ // refresh job
+ LOG.warn(
+ "Failed to start the continuous refresh job for materialized table {} using new query {}, rollback to origin query {}.",
+ tableIdentifier,
+ op.getNewMaterializedTable().getDefinitionQuery(),
+ suspendMaterializedTable.getDefinitionQuery(),
+ e);
+
+ AlterMaterializedTableChangeOperation rollbackChangeOperation =
+ generateRollbackAlterMaterializedTableOperation(
+ suspendMaterializedTable, alterMaterializedTableChangeOperation);
+ operationExecutor.callExecutableOperation(handle, rollbackChangeOperation);
+
+ ContinuousRefreshHandler continuousRefreshHandler =
+ deserializeContinuousHandler(
+ suspendMaterializedTable.getSerializedRefreshHandler());
+ executeContinuousRefreshJob(
+ operationExecutor,
+ handle,
+ suspendMaterializedTable,
+ tableIdentifier,
+ Collections.emptyMap(),
+ continuousRefreshHandler.getRestorePath());
+
+ throw new SqlExecutionException(
+ String.format(
+ "Failed to start the continuous refresh job using new query %s when altering materialized table %s select query.",
+ op.getNewMaterializedTable().getDefinitionQuery(), tableIdentifier),
+ e);
+ }
+ } else if (CatalogMaterializedTable.RefreshStatus.SUSPENDED
+ == oldMaterializedTable.getRefreshStatus()) {
+ // alter schema & definition query & refresh handler (reset savepoint path of refresh
+ // handler)
+ List<MaterializedTableChange> tableChanges = new ArrayList<>(op.getTableChanges());
+ TableChange.ModifyRefreshHandler modifyRefreshHandler =
+ genereateResetSavepointTableChange(
+ oldMaterializedTable.getSerializedRefreshHandler());
+ tableChanges.add(modifyRefreshHandler);
+
+ CatalogMaterializedTable updatedMaterializedTable =
+ op.getNewMaterializedTable()
+ .copy(
+ oldMaterializedTable.getRefreshStatus(),
+ modifyRefreshHandler.getRefreshHandlerDesc(),
+ modifyRefreshHandler.getRefreshHandlerBytes());
+ AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
+ new AlterMaterializedTableChangeOperation(
+ tableIdentifier, tableChanges, updatedMaterializedTable);
+
+ operationExecutor.callExecutableOperation(
+ handle, alterMaterializedTableChangeOperation);
+ } else {
+ throw new SqlExecutionException(
+ String.format(
+ "Materialized table %s is being initialized and does not support alter operation.",
+ tableIdentifier));
+ }
+
+ return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
+ }
+
+ private AlterMaterializedTableChangeOperation generateRollbackAlterMaterializedTableOperation(
+ CatalogMaterializedTable oldMaterializedTable,
+ AlterMaterializedTableChangeOperation op) {
+ List<MaterializedTableChange> tableChanges = op.getTableChanges();
+ List<MaterializedTableChange> rollbackChanges = new ArrayList<>();
+
+ for (TableChange tableChange : tableChanges) {
+ if (tableChange instanceof TableChange.AddColumn) {
+ TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange;
+ rollbackChanges.add(TableChange.dropColumn(addColumn.getColumn().getName()));
+ } else if (tableChange instanceof TableChange.ModifyRefreshHandler) {
+ rollbackChanges.add(
+ TableChange.modifyRefreshHandler(
+ oldMaterializedTable.getRefreshHandlerDescription().orElse(null),
+ oldMaterializedTable.getSerializedRefreshHandler()));
+ } else if (tableChange instanceof TableChange.ModifyDefinitionQuery) {
+ rollbackChanges.add(
+ TableChange.modifyDefinitionQuery(
+ oldMaterializedTable.getDefinitionQuery()));
+ } else {
+ throw new ValidationException(
+ String.format(
+ "Failed to generate rollback changes for materialized table '%s'. "
+ + "Unsupported table change detected: %s. ",
+ op.getTableIdentifier(), tableChange));
+ }
+ }
+
+ return new AlterMaterializedTableChangeOperation(
+ op.getTableIdentifier(), rollbackChanges, oldMaterializedTable);
+ }
+
+ private TableChange.ModifyRefreshHandler genereateResetSavepointTableChange(
+ byte[] serializedContinuousHandler) {
+ ContinuousRefreshHandler continuousRefreshHandler =
+ deserializeContinuousHandler(serializedContinuousHandler);
+ ContinuousRefreshHandler resetedRefreshHandler =
+ new ContinuousRefreshHandler(
+ continuousRefreshHandler.getExecutionTarget(),
+ continuousRefreshHandler.getJobId());
+
+ return TableChange.modifyRefreshHandler(
+ resetedRefreshHandler.asSummaryString(),
+ serializeContinuousHandler(resetedRefreshHandler));
+ }
+
private ResultFetcher callDropMaterializedTableOperation(
OperationExecutor operationExecutor,
OperationHandle handle,
@@ -1018,7 +1181,7 @@ public class MaterializedTableManager {
return (ResolvedCatalogMaterializedTable) resolvedCatalogBaseTable;
}
- private void updateRefreshHandler(
+ private CatalogMaterializedTable updateRefreshHandler(
OperationExecutor operationExecutor,
OperationHandle operationHandle,
ObjectIdentifier materializedTableIdentifier,
@@ -1029,7 +1192,7 @@ public class MaterializedTableManager {
CatalogMaterializedTable updatedMaterializedTable =
catalogMaterializedTable.copy(
refreshStatus, refreshHandlerSummary, serializedRefreshHandler);
- List<TableChange> tableChanges = new ArrayList<>();
+ List<MaterializedTableChange> tableChanges = new ArrayList<>();
tableChanges.add(TableChange.modifyRefreshStatus(refreshStatus));
tableChanges.add(
TableChange.modifyRefreshHandler(refreshHandlerSummary, serializedRefreshHandler));
@@ -1039,6 +1202,8 @@ public class MaterializedTableManager {
// update RefreshHandler to Catalog
operationExecutor.callExecutableOperation(
operationHandle, alterMaterializedTableChangeOperation);
+
+ return updatedMaterializedTable;
}
/** Generate insert statement for materialized table. */
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 b5f5fc3bd5f..49754e42aad 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
@@ -72,6 +72,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import static org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_INTERVAL;
import static org.apache.flink.table.api.config.TableConfigOptions.RESOURCES_DOWNLOAD_DIR;
@@ -1027,6 +1028,348 @@ public class MaterializedTableStatementITCase extends AbstractMaterializedTableS
fileSystemCatalogName,
TEST_DEFAULT_DATABASE,
"users_shops")));
+
+ dropMaterializedTable(
+ ObjectIdentifier.of(fileSystemCatalogName, TEST_DEFAULT_DATABASE, "users_shops"));
+ }
+
+ @Test
+ void testAlterMaterializedTableAsQueryInFullMode() throws Exception {
+ createAndVerifyCreateMaterializedTableWithData(
+ "users_shops", Collections.emptyList(), Collections.emptyMap(), RefreshMode.FULL);
+
+ ResolvedCatalogMaterializedTable oldTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ // Alter materialized table as query in full mode
+ String alterMaterializedTableAsQueryDDL =
+ "ALTER MATERIALIZED TABLE users_shops"
+ + " AS SELECT \n"
+ + " user_id,\n"
+ + " shop_id,\n"
+ + " ds,\n"
+ + " COUNT(order_id) AS order_cnt,\n"
+ + " SUM(order_amount) AS order_amount_sum\n"
+ + " FROM (\n"
+ + " SELECT user_id, shop_id, order_created_at AS ds, order_id, 1 as order_amount FROM my_source"
+ + " ) AS tmp\n"
+ + " GROUP BY (user_id, shop_id, ds)";
+
+ OperationHandle alterMaterializedTableAsQueryHandle =
+ service.executeStatement(
+ sessionHandle, alterMaterializedTableAsQueryDDL, -1, new Configuration());
+
+ awaitOperationTermination(service, sessionHandle, alterMaterializedTableAsQueryHandle);
+
+ // verify the altered materialized table
+ ResolvedCatalogMaterializedTable newTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema()))
+ .isEqualTo(
+ Collections.singletonList(
+ Column.physical("order_amount_sum", DataTypes.INT())));
+ assertThat(newTable.getDefinitionQuery())
+ .isEqualTo(
+ String.format(
+ "SELECT `tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`, COUNT(`tmp`.`order_id`) AS `order_cnt`, SUM(`tmp`.`order_amount`) AS `order_amount_sum`\n"
+ + "FROM (SELECT `my_source`.`user_id`, `my_source`.`shop_id`, `my_source`.`order_created_at` AS `ds`, `my_source`.`order_id`, 1 AS `order_amount`\n"
+ + "FROM `%s`.`test_db`.`my_source`) AS `tmp`\n"
+ + "GROUP BY ROW(`tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`)",
+ fileSystemCatalogName));
+ // the refresh handler in full mode should be the same as the old one
+ assertThat(oldTable.getSerializedRefreshHandler())
+ .isEqualTo(newTable.getSerializedRefreshHandler());
+ assertThat(oldTable.getDefinitionFreshness()).isEqualTo(newTable.getDefinitionFreshness());
+ }
+
+ @Test
+ void testAlterMaterializedTableAsQueryInFullModeWithSuspendStatus() throws Exception {
+ createAndVerifyCreateMaterializedTableWithData(
+ "users_shops", Collections.emptyList(), Collections.emptyMap(), RefreshMode.FULL);
+
+ ResolvedCatalogMaterializedTable oldTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ // Alter materialized table suspend
+ String alterMaterializedTableSuspendDDL = "ALTER MATERIALIZED TABLE users_shops SUSPEND";
+ OperationHandle alterMaterializedTableSuspendHandle =
+ service.executeStatement(
+ sessionHandle, alterMaterializedTableSuspendDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, alterMaterializedTableSuspendHandle);
+
+ // Alter materialized table as query in full mode
+ String alterMaterializedTableAsQueryDDL =
+ "ALTER MATERIALIZED TABLE users_shops"
+ + " AS SELECT \n"
+ + " user_id,\n"
+ + " shop_id,\n"
+ + " ds,\n"
+ + " COUNT(order_id) AS order_cnt,\n"
+ + " SUM(order_amount) AS order_amount_sum\n"
+ + " FROM (\n"
+ + " SELECT user_id, shop_id, order_created_at AS ds, order_id, 1 as order_amount FROM my_source"
+ + " ) AS tmp\n"
+ + " GROUP BY (user_id, shop_id, ds)";
+
+ OperationHandle alterMaterializedTableAsQueryHandle =
+ service.executeStatement(
+ sessionHandle, alterMaterializedTableAsQueryDDL, -1, new Configuration());
+
+ awaitOperationTermination(service, sessionHandle, alterMaterializedTableAsQueryHandle);
+
+ // verify the altered materialized table
+ ResolvedCatalogMaterializedTable newTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema()))
+ .isEqualTo(
+ Collections.singletonList(
+ Column.physical("order_amount_sum", DataTypes.INT())));
+ assertThat(newTable.getDefinitionQuery())
+ .isEqualTo(
+ String.format(
+ "SELECT `tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`, COUNT(`tmp`.`order_id`) AS `order_cnt`, SUM(`tmp`.`order_amount`) AS `order_amount_sum`\n"
+ + "FROM (SELECT `my_source`.`user_id`, `my_source`.`shop_id`, `my_source`.`order_created_at` AS `ds`, `my_source`.`order_id`, 1 AS `order_amount`\n"
+ + "FROM `%s`.`test_db`.`my_source`) AS `tmp`\n"
+ + "GROUP BY ROW(`tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`)",
+ fileSystemCatalogName));
+
+ // the refresh handler in full mode should be the same as the old one
+ assertThat(oldTable.getSerializedRefreshHandler())
+ .isEqualTo(newTable.getSerializedRefreshHandler());
+ assertThat(oldTable.getDefinitionFreshness()).isEqualTo(newTable.getDefinitionFreshness());
+ }
+
+ @Test
+ void testAlterMaterializedTableAsQueryInContinuousMode(@TempDir Path temporaryPath)
+ throws Exception {
+ String materializedTableDDL =
+ "CREATE MATERIALIZED TABLE users_shops ("
+ + " PRIMARY KEY (ds, user_id) not enforced)"
+ + " PARTITIONED BY (ds)\n"
+ + " WITH(\n"
+ + " 'format' = 'debezium-json'\n"
+ + " )\n"
+ + " FRESHNESS = INTERVAL '30' SECOND\n"
+ + " AS SELECT \n"
+ + " coalesce(user_id, 0) as user_id,\n"
+ + " shop_id,\n"
+ + " coalesce(ds, '') as ds,\n"
+ + " SUM (payment_amount_cents) AS payed_buy_fee_sum\n"
+ + " FROM (\n"
+ + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource"
+ + " ) AS tmp\n"
+ + " GROUP BY (user_id, shop_id, ds)";
+ OperationHandle materializedTableHandle =
+ service.executeStatement(
+ sessionHandle, materializedTableDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, materializedTableHandle);
+ ResolvedCatalogMaterializedTable oldTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ // verify background job is running
+ ContinuousRefreshHandler activeRefreshHandler =
+ ContinuousRefreshHandlerSerializer.INSTANCE.deserialize(
+ oldTable.getSerializedRefreshHandler(), getClass().getClassLoader());
+ waitUntilAllTasksAreRunning(
+ restClusterClient, JobID.fromHexString(activeRefreshHandler.getJobId()));
+
+ // setup savepoint dir
+ String savepointDir = "file://" + temporaryPath.toAbsolutePath();
+ String setupSavepointDDL =
+ "SET 'execution.checkpointing.savepoint-dir' = '" + savepointDir + "'";
+ OperationHandle setupSavepointHandle =
+ service.executeStatement(sessionHandle, setupSavepointDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, setupSavepointHandle);
+
+ String alterTableDDL =
+ "ALTER MATERIALIZED TABLE users_shops"
+ + " AS SELECT \n"
+ + " coalesce(user_id, 0) as user_id,\n"
+ + " shop_id,\n"
+ + " coalesce(ds, '') as ds,\n"
+ + " SUM (payment_amount_cents) AS payed_buy_fee_sum,\n"
+ + " SUM (1) AS pv\n"
+ + " FROM (\n"
+ + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource"
+ + " ) AS tmp\n"
+ + " GROUP BY (user_id, shop_id, ds)";
+ OperationHandle alterTableHandle =
+ service.executeStatement(sessionHandle, alterTableDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, alterTableHandle);
+ ResolvedCatalogMaterializedTable newTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema()))
+ .isEqualTo(Collections.singletonList(Column.physical("pv", DataTypes.INT())));
+ assertThat(newTable.getResolvedSchema().getPrimaryKey())
+ .isEqualTo(oldTable.getResolvedSchema().getPrimaryKey());
+ assertThat(newTable.getResolvedSchema().getWatermarkSpecs())
+ .isEqualTo(oldTable.getResolvedSchema().getWatermarkSpecs());
+ assertThat(newTable.getDefinitionQuery())
+ .isEqualTo(
+ String.format(
+ "SELECT COALESCE(`tmp`.`user_id`, 0) AS `user_id`, `tmp`.`shop_id`, COALESCE(`tmp`.`ds`, '') AS `ds`, SUM(`tmp`.`payment_amount_cents`) AS `payed_buy_fee_sum`, SUM(1) AS `pv`\n"
+ + "FROM (SELECT `datagenSource`.`user_id`, `datagenSource`.`shop_id`, `DATE_FORMAT`(`datagenSource`.`order_created_at`, 'yyyy-MM-dd') AS `ds`, `datagenSource`.`payment_amount_cents`\n"
+ + "FROM `%s`.`test_db`.`datagenSource`) AS `tmp`\n"
+ + "GROUP BY ROW(`tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`)",
+ fileSystemCatalogName));
+ assertThat(oldTable.getSerializedRefreshHandler())
+ .isNotEqualTo(newTable.getSerializedRefreshHandler());
+
+ // verify the new continuous job is start without savepoint
+ ContinuousRefreshHandler newContinuousRefreshHandler =
+ ContinuousRefreshHandlerSerializer.INSTANCE.deserialize(
+ newTable.getSerializedRefreshHandler(),
+ Thread.currentThread().getContextClassLoader());
+ Optional<String> restorePath =
+ getJobRestoreSavepointPath(
+ restClusterClient, newContinuousRefreshHandler.getJobId());
+ assertThat(restorePath).isEmpty();
+
+ // drop the materialized table
+ dropMaterializedTable(
+ ObjectIdentifier.of(fileSystemCatalogName, TEST_DEFAULT_DATABASE, "users_shops"));
+ }
+
+ @Test
+ void testAlterMaterializedTableAsQueryInContinuousModeWithSuspendStatus(
+ @TempDir Path temporaryPath) throws Exception {
+ String materializedTableDDL =
+ "CREATE MATERIALIZED TABLE users_shops"
+ + " PARTITIONED BY (ds)\n"
+ + " WITH(\n"
+ + " 'format' = 'debezium-json'\n"
+ + " )\n"
+ + " FRESHNESS = INTERVAL '30' SECOND\n"
+ + " AS SELECT \n"
+ + " user_id,\n"
+ + " shop_id,\n"
+ + " ds,\n"
+ + " SUM (payment_amount_cents) AS payed_buy_fee_sum\n"
+ + " FROM (\n"
+ + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource"
+ + " ) AS tmp\n"
+ + " GROUP BY (user_id, shop_id, ds)";
+ OperationHandle materializedTableHandle =
+ service.executeStatement(
+ sessionHandle, materializedTableDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, materializedTableHandle);
+ ResolvedCatalogMaterializedTable oldTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ // verify background job is running
+ ContinuousRefreshHandler activeRefreshHandler =
+ ContinuousRefreshHandlerSerializer.INSTANCE.deserialize(
+ oldTable.getSerializedRefreshHandler(), getClass().getClassLoader());
+ waitUntilAllTasksAreRunning(
+ restClusterClient, JobID.fromHexString(activeRefreshHandler.getJobId()));
+
+ // setup savepoint dir
+ String savepointDir = "file://" + temporaryPath.toAbsolutePath();
+ String setupSavepointDDL =
+ "SET 'execution.checkpointing.savepoint-dir' = '" + savepointDir + "'";
+ OperationHandle setupSavepointHandle =
+ service.executeStatement(sessionHandle, setupSavepointDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, setupSavepointHandle);
+
+ // suspend materialized table
+ String suspendTableDDL = "ALTER MATERIALIZED TABLE users_shops SUSPEND";
+ OperationHandle suspendTableHandle =
+ service.executeStatement(sessionHandle, suspendTableDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, suspendTableHandle);
+
+ String alterTableDDL =
+ "ALTER MATERIALIZED TABLE users_shops"
+ + " AS SELECT \n"
+ + " user_id,\n"
+ + " shop_id,\n"
+ + " ds,\n"
+ + " SUM (payment_amount_cents) AS payed_buy_fee_sum,\n"
+ + " SUM (1) AS pv\n"
+ + " FROM (\n"
+ + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource"
+ + " ) AS tmp\n"
+ + " GROUP BY (user_id, shop_id, ds)";
+ OperationHandle alterTableHandle =
+ service.executeStatement(sessionHandle, alterTableDDL, -1, new Configuration());
+ awaitOperationTermination(service, sessionHandle, alterTableHandle);
+ ResolvedCatalogMaterializedTable newTable =
+ (ResolvedCatalogMaterializedTable)
+ service.getTable(
+ sessionHandle,
+ ObjectIdentifier.of(
+ fileSystemCatalogName,
+ TEST_DEFAULT_DATABASE,
+ "users_shops"));
+
+ assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema()))
+ .isEqualTo(Collections.singletonList(Column.physical("pv", DataTypes.INT())));
+ assertThat(newTable.getDefinitionQuery())
+ .isEqualTo(
+ String.format(
+ "SELECT `tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`, SUM(`tmp`.`payment_amount_cents`) AS `payed_buy_fee_sum`, SUM(1) AS `pv`\n"
+ + "FROM (SELECT `datagenSource`.`user_id`, `datagenSource`.`shop_id`, `DATE_FORMAT`(`datagenSource`.`order_created_at`, 'yyyy-MM-dd') AS `ds`, `datagenSource`.`payment_amount_cents`\n"
+ + "FROM `%s`.`test_db`.`datagenSource`) AS `tmp`\n"
+ + "GROUP BY ROW(`tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`)",
+ fileSystemCatalogName));
+ assertThat(oldTable.getSerializedRefreshHandler())
+ .isEqualTo(newTable.getSerializedRefreshHandler());
+ assertThat(oldTable.getDefinitionFreshness()).isEqualTo(newTable.getDefinitionFreshness());
+
+ // verify the restore path is empty
+ ContinuousRefreshHandler newContinuousRefreshHandler =
+ ContinuousRefreshHandlerSerializer.INSTANCE.deserialize(
+ newTable.getSerializedRefreshHandler(),
+ Thread.currentThread().getContextClassLoader());
+ assertThat(newContinuousRefreshHandler.getRestorePath()).isEmpty();
+
+ // drop the materialized table
+ dropMaterializedTable(
+ ObjectIdentifier.of(fileSystemCatalogName, TEST_DEFAULT_DATABASE, "users_shops"));
}
@Test
@@ -1574,6 +1917,12 @@ public class MaterializedTableStatementITCase extends AbstractMaterializedTableS
fileSystemCatalogName, TEST_DEFAULT_DATABASE, "my_materialized_table"));
}
+ private List<Column> getAddedColumns(ResolvedSchema newSchema, ResolvedSchema oldSchema) {
+ return newSchema.getColumns().stream()
+ .filter(column -> !oldSchema.getColumns().contains(column))
+ .collect(Collectors.toList());
+ }
+
private int getPartitionSize(List<Row> data, String partition) {
return (int)
data.stream().filter(row -> row.getField(3).toString().equals(partition)).count();
(flink) 04/04: [FLINK-36994][table] Fix expand sql error when materialized table creation with UDTF queries
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5e0e5a233eb6f605998d5e64d182203d172f0797
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Wed Jan 8 09:52:05 2025 +0800
[FLINK-36994][table] Fix expand sql error when materialized table creation with UDTF queries
---
.../SqlCreateMaterializedTableConverter.java | 14 +++++---
...erializedTableNodeToOperationConverterTest.java | 42 ++++++++++++++++++++--
2 files changed, 49 insertions(+), 7 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java
index de14add9692..df30527cb5c 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateMaterializedTableConverter.java
@@ -116,14 +116,20 @@ public class SqlCreateMaterializedTableConverter
}
// get query schema and definition query
- SqlNode validateQuery =
- context.getSqlValidator().validate(sqlCreateMaterializedTable.getAsQuery());
+ SqlNode selectQuery = sqlCreateMaterializedTable.getAsQuery();
+ String originalQuery = context.toQuotedSqlString(selectQuery);
+ SqlNode validateQuery = context.getSqlValidator().validate(selectQuery);
+
+ // The LATERAL operator was eliminated during sql validation, thus the unparsed SQL
+ // does not contain LATERAL which is problematic,
+ // the issue was resolved in CALCITE-4077
+ // (always treat the table function as implicitly LATERAL).
+ String definitionQuery = context.expandSqlIdentifiers(originalQuery);
+
PlannerQueryOperation queryOperation =
new PlannerQueryOperation(
context.toRelRoot(validateQuery).project(),
() -> context.toQuotedSqlString(validateQuery));
- String definitionQuery =
- context.expandSqlIdentifiers(queryOperation.asSerializableString());
// get schema
ResolvedSchema resolvedSchema = queryOperation.getResolvedSchema();
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 a3b659f335b..63a91f7e533 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
@@ -25,10 +25,12 @@ import org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.IntervalFreshness;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
@@ -39,6 +41,7 @@ import org.apache.flink.table.operations.materializedtable.AlterMaterializedTabl
import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableSuspendOperation;
import org.apache.flink.table.operations.materializedtable.CreateMaterializedTableOperation;
import org.apache.flink.table.operations.materializedtable.DropMaterializedTableOperation;
+import org.apache.flink.table.planner.utils.TableFunc0;
import org.apache.flink.shaded.guava32.com.google.common.collect.ImmutableMap;
@@ -140,15 +143,48 @@ public class SqlMaterializedTableNodeToOperationConverterTest
.logicalRefreshMode(CatalogMaterializedTable.LogicalRefreshMode.FULL)
.refreshMode(CatalogMaterializedTable.RefreshMode.FULL)
.refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
- .definitionQuery(
- "SELECT `t1`.`a`, `t1`.`b`, `t1`.`c`, `t1`.`d`\n"
- + "FROM `builtin`.`default`.`t1` AS `t1`")
+ .definitionQuery("SELECT *\n" + "FROM `builtin`.`default`.`t1`")
.build();
assertThat(((ResolvedCatalogMaterializedTable) materializedTable).getOrigin())
.isEqualTo(expected);
}
+ @Test
+ void testCreateMaterializedTableWithUDTFQuery() {
+ functionCatalog.registerCatalogFunction(
+ UnresolvedIdentifier.of(
+ ObjectIdentifier.of(
+ catalogManager.getCurrentCatalog(), "default", "myFunc")),
+ TableFunc0.class,
+ true);
+
+ final String sql =
+ "CREATE MATERIALIZED TABLE mtbl1 (\n"
+ + " CONSTRAINT ct1 PRIMARY KEY(a) NOT ENFORCED"
+ + ")\n"
+ + "COMMENT 'materialized table comment'\n"
+ + "PARTITIONED BY (a)\n"
+ + "WITH (\n"
+ + " 'connector' = 'filesystem', \n"
+ + " 'format' = 'json'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '30' SECOND\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS SELECT a, f1, f2 FROM t1,LATERAL TABLE(myFunc(b)) as T(f1, f2)";
+ Operation operation = parse(sql);
+ assertThat(operation).isInstanceOf(CreateMaterializedTableOperation.class);
+
+ CreateMaterializedTableOperation createOperation =
+ (CreateMaterializedTableOperation) operation;
+
+ assertThat(createOperation.getCatalogMaterializedTable().getDefinitionQuery())
+ .isEqualTo(
+ "SELECT `t1`.`a`, `T`.`f1`, `T`.`f2`\n"
+ + "FROM `builtin`.`default`.`t1`,\n"
+ + "LATERAL TABLE(`builtin`.`default`.`myFunc`(`b`)) AS `T` (`f1`, `f2`)");
+ }
+
@Test
void testContinuousRefreshMode() {
// test continuous mode derived by specify freshness automatically