This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 a6669809b9f [FLINK-39957][table] Support `EXPLAIN` of `CREATE [OR
ALTER] MATERIALIZED TABLE`
a6669809b9f is described below
commit a6669809b9f54bf1fbee2615e3f4c1bc28a002ed
Author: Ramin Gharib <[email protected]>
AuthorDate: Sun Jul 5 09:30:36 2026 +0200
[FLINK-39957][table] Support `EXPLAIN` of `CREATE [OR ALTER] MATERIALIZED
TABLE`
---
.../MaterializedTableManager.java | 13 +-
.../service/operation/OperationExecutor.java | 5 +-
.../AbstractMaterializedTableStatementITCase.java | 44 ++--
.../service/MaterializedTableConversionITCase.java | 39 +++
.../service/MaterializedTableStatementITCase.java | 290 ++++++++++++++-------
.../src/main/codegen/includes/parserImpls.ftl | 11 +-
.../flink/sql/parser/utils/ParserResource.java | 4 +
.../flink/sql/parser/FlinkSqlParserImplTest.java | 48 ++--
.../table/operations/ModifyOperationVisitor.java | 9 +
.../AlterMaterializedTableAsQueryOperation.java | 6 +-
.../AlterMaterializedTableChangeOperation.java | 33 ++-
.../ConvertTableToMaterializedTableOperation.java | 24 +-
.../CreateMaterializedTableOperation.java | 32 ++-
.../FullAlterMaterializedTableOperation.java | 6 +-
...ializedTableAsQueryOperationValidationTest.java | 2 +-
.../table/planner/connectors/DynamicSinkUtils.java | 42 ++-
.../operations/SqlNodeToOperationConversion.java | 8 +-
.../operations/converters/SqlNodeConvertUtils.java | 3 +-
.../SqlAlterMaterializedTableAsQueryConverter.java | 10 +-
...SqlCreateOrAlterMaterializedTableConverter.java | 44 +++-
.../table/planner/delegation/PlannerBase.scala | 68 ++++-
...erializedTableNodeToOperationConverterTest.java | 42 +++
.../table/planner/plan/stream/sql/ExplainTest.java | 203 +++++++++++++++
.../explain/testExplainAlterMaterializedTable.out | 12 +
.../testExplainConvertTableToMaterializedTable.out | 12 +
.../explain/testExplainCreateMaterializedTable.out | 12 +
.../testExplainCreateOrAlterMaterializedTable.out | 14 +
.../testExplainFullAlterMaterializedTable.out | 14 +
28 files changed, 883 insertions(+), 167 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 02c646440b1..56cf110880a 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
@@ -954,7 +954,8 @@ public class MaterializedTableManager {
new AlterMaterializedTableChangeOperation(
op.getTableIdentifier(),
oldTable -> op.getTableChanges(),
- suspendMaterializedTable);
+ suspendMaterializedTable,
+ op.getSinkModifyQuery());
operationExecutor.callExecutableOperation(
handle, alterMaterializedTableChangeOperation);
@@ -1010,7 +1011,10 @@ public class MaterializedTableManager {
AlterMaterializedTableChangeOperation
alterMaterializedTableChangeOperation =
new AlterMaterializedTableChangeOperation(
- tableIdentifier, oldTable -> tableChanges,
oldMaterializedTable);
+ tableIdentifier,
+ oldTable -> tableChanges,
+ oldMaterializedTable,
+ op.getSinkModifyQuery());
operationExecutor.callExecutableOperation(
handle, alterMaterializedTableChangeOperation);
@@ -1029,7 +1033,10 @@ public class MaterializedTableManager {
AlterMaterializedTableChangeOperation op) {
return new AlterMaterializedTableChangeOperation(
- op.getTableIdentifier(), oldTable -> List.of(),
oldMaterializedTable);
+ op.getTableIdentifier(),
+ oldTable -> List.of(),
+ oldMaterializedTable,
+ op.getSinkModifyQuery());
}
private TableChange.ModifyRefreshHandler generateResetSavepointTableChange(
diff --git
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
index 5735abd6dee..2bfd11eadec 100644
---
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
+++
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
@@ -494,7 +494,8 @@ public class OperationExecutor {
TableEnvironmentInternal tableEnv, OperationHandle handle,
Operation operation) {
if (operation instanceof EndStatementSetOperation) {
return callEndStatementSetOperation(tableEnv, handle);
- } else if (operation instanceof ModifyOperation) {
+ } else if (operation instanceof ModifyOperation
+ && !(operation instanceof MaterializedTableOperation)) {
sessionContext.addStatementSetOperation((ModifyOperation)
operation);
return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK,
false);
} else {
@@ -518,7 +519,7 @@ public class OperationExecutor {
} else if (op instanceof EndStatementSetOperation) {
throw new SqlExecutionException(
"No Statement Set to submit. 'END' statement should be
used after 'BEGIN STATEMENT SET'.");
- } else if (op instanceof ModifyOperation) {
+ } else if (op instanceof ModifyOperation && !(op instanceof
MaterializedTableOperation)) {
return callModifyOperations(
tableEnv, handle,
Collections.singletonList((ModifyOperation) op));
} else if (op instanceof CompileAndExecutePlanOperation
diff --git
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractMaterializedTableStatementITCase.java
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractMaterializedTableStatementITCase.java
index 7ec2aad15de..d7e95fbac36 100644
---
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractMaterializedTableStatementITCase.java
+++
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractMaterializedTableStatementITCase.java
@@ -221,24 +221,7 @@ public abstract class
AbstractMaterializedTableStatementITCase {
long timeout = Duration.ofSeconds(20).toMillis();
long pause = Duration.ofSeconds(2).toMillis();
- String dataId = TestValuesTableFactory.registerData(data);
- String sourceDdl =
- String.format(
- "CREATE TABLE IF NOT EXISTS my_source (\n"
- + " order_id BIGINT,\n"
- + " user_id BIGINT,\n"
- + " shop_id BIGINT,\n"
- + " order_created_at STRING\n"
- + ")\n"
- + "WITH (\n"
- + " 'connector' = 'values',\n"
- + " 'bounded' = 'true',\n"
- + " 'data-id' = '%s'\n"
- + ")",
- dataId);
- OperationHandle sourceHandle =
- service.executeStatement(sessionHandle, sourceDdl, -1, new
Configuration());
- awaitOperationTermination(service, sessionHandle, sourceHandle);
+ createBoundedValuesSource(data);
String partitionFields =
partitionFormatter != null && !partitionFormatter.isEmpty()
@@ -290,6 +273,28 @@ public abstract class
AbstractMaterializedTableStatementITCase {
"Failed to verify the data in materialized table.");
}
+ /** Registers a bounded {@code my_source} table backed by empty values
data. */
+ public void createBoundedValuesSource(List<Row> data) throws Exception {
+ String dataId = TestValuesTableFactory.registerData(data);
+ String sourceDDL =
+ String.format(
+ "CREATE TABLE IF NOT EXISTS my_source (\n"
+ + " order_id BIGINT,\n"
+ + " user_id BIGINT,\n"
+ + " shop_id BIGINT,\n"
+ + " order_created_at STRING\n"
+ + ")\n"
+ + "WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'true',\n"
+ + " 'data-id' = '%s'\n"
+ + ")",
+ dataId);
+ OperationHandle sourceHandle =
+ service.executeStatement(sessionHandle, sourceDDL, -1, new
Configuration());
+ awaitOperationTermination(service, sessionHandle, sourceHandle);
+ }
+
public List<RowData> fetchTableData(SessionHandle sessionHandle, String
query) {
OperationHandle queryHandle =
service.executeStatement(sessionHandle, query, -1, new
Configuration());
@@ -337,7 +342,8 @@ public abstract class
AbstractMaterializedTableStatementITCase {
public void dropMaterializedTable(ObjectIdentifier objectIdentifier)
throws Exception {
String dropMaterializedTableDDL =
String.format(
- "DROP MATERIALIZED TABLE %s",
objectIdentifier.asSerializableString());
+ "DROP MATERIALIZED TABLE IF EXISTS %s",
+ objectIdentifier.asSerializableString());
OperationHandle dropMaterializedTableHandle =
service.executeStatement(
sessionHandle, dropMaterializedTableDDL, -1, new
Configuration());
diff --git
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableConversionITCase.java
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableConversionITCase.java
index 870648e9e96..ce9285f70db 100644
---
a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableConversionITCase.java
+++
b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableConversionITCase.java
@@ -57,6 +57,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
+import static org.apache.flink.table.catalog.CatalogBaseTable.TableKind.TABLE;
import static
org.apache.flink.table.catalog.CommonCatalogOptions.TABLE_CATALOG_STORE_KIND;
import static
org.apache.flink.table.factories.FactoryUtil.WORKFLOW_SCHEDULER_TYPE;
import static
org.apache.flink.table.gateway.service.MaterializedTableTestUtils.executeStatement;
@@ -293,6 +294,44 @@ class MaterializedTableConversionITCase {
assertThat(suspendedMaterializedTable.getRefreshStatus()).isSameAs(RefreshStatus.SUSPENDED);
}
+ @Test
+ void testExplainConvertTableToMaterializedTable() throws Exception {
+ // pre-create a regular table that CREATE OR ALTER would convert in
place.
+ String createRegularTableDDL =
+ "CREATE TABLE users_shops (\n"
+ + " user_id BIGINT,\n"
+ + " shop_id BIGINT,\n"
+ + " payment_amount_cents BIGINT\n"
+ + ")\n"
+ + "WITH ('connector' = 'values')";
+ OperationHandle createTableHandle =
+ executeStatement(service, sessionHandle,
createRegularTableDDL);
+ awaitOperationTermination(service, sessionHandle, createTableHandle);
+
+ String explainDDL =
+ "EXPLAIN CREATE OR ALTER MATERIALIZED TABLE users_shops"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " AS SELECT user_id, shop_id, payment_amount_cents
FROM datagenSource";
+
+ OperationHandle explainHandle = executeStatement(service,
sessionHandle, explainDDL);
+ awaitOperationTermination(service, sessionHandle, explainHandle);
+ List<RowData> explainResults = fetchAllResults(service, sessionHandle,
explainHandle);
+ assertThat(explainResults).hasSize(1);
+ assertThat(explainResults.get(0).getString(0).toString())
+ .contains(
+ "== Abstract Syntax Tree ==",
+ "== Optimized Physical Plan ==",
+ "== Optimized Execution Plan ==",
+ "Sink(table=[",
+ "users_shops");
+
+ // EXPLAIN is side-effect-free: the entry is still a regular table,
not converted.
+ assertThat(
+ service.getTable(sessionHandle,
getObjectIdentifier("users_shops"))
+ .getTableKind())
+ .isSameAs(TABLE);
+ }
+
private SessionHandle initializeSession() {
SessionHandle sessionHandle =
service.openSession(defaultSessionEnvironment);
String catalogDDL =
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 134015c565d..ddc0fcd8500 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
@@ -54,11 +54,11 @@ import
org.apache.flink.table.gateway.workflow.EmbeddedRefreshHandler;
import
org.apache.flink.table.gateway.workflow.EmbeddedRefreshHandlerSerializer;
import org.apache.flink.table.gateway.workflow.WorkflowInfo;
import
org.apache.flink.table.gateway.workflow.scheduler.EmbeddedQuartzScheduler;
-import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.table.refresh.ContinuousRefreshHandler;
import org.apache.flink.table.refresh.ContinuousRefreshHandlerSerializer;
import org.apache.flink.types.Row;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.quartz.JobDetail;
@@ -99,6 +99,12 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
*/
class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatementITCase {
+ @AfterEach
+ void tearDown() throws Exception {
+ ObjectIdentifier userShopsIdentifier =
getObjectIdentifier("users_shops");
+ dropMaterializedTable(userShopsIdentifier);
+ }
+
@Test
void testCreateMaterializedTableInContinuousMode() throws Exception {
String materializedTableDDL =
@@ -161,9 +167,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
long checkpointInterval =
getCheckpointIntervalConfig(restClusterClient,
activeRefreshHandler.getJobId());
assertThat(checkpointInterval).isEqualTo(30 * 1000);
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -244,9 +247,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
.containsKey("table.catalog-store.file.path")
.doesNotContainKey(WORKFLOW_SCHEDULER_TYPE.key())
.doesNotContainKey(RESOURCES_DOWNLOAD_DIR.key());
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -316,9 +316,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
// default checkpoint interval is 3 minutes
final int expectedCheckpointInterval = 60 * 3 * 1000;
assertThat(checkpointInterval).isEqualTo(expectedCheckpointInterval);
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -370,31 +367,11 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
long actualCheckpointInterval =
getCheckpointIntervalConfig(restClusterClient,
activeRefreshHandler.getJobId());
assertThat(actualCheckpointInterval).isEqualTo(checkpointInterval);
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
void testCreateMaterializedTableInFullMode() throws Exception {
- String dataId = TestValuesTableFactory.registerData(List.of());
- String sourceDdl =
- String.format(
- "CREATE TABLE IF NOT EXISTS my_source (\n"
- + " order_id BIGINT,\n"
- + " user_id BIGINT,\n"
- + " shop_id BIGINT,\n"
- + " order_created_at STRING\n"
- + ")\n"
- + "WITH (\n"
- + " 'connector' = 'values',\n"
- + " 'bounded' = 'true',\n"
- + " 'data-id' = '%s'\n"
- + ")",
- dataId);
-
- OperationHandle sourceHandle = executeStatement(sourceDdl);
- awaitOperationTermination(service, sessionHandle, sourceHandle);
+ createBoundedValuesSource(List.of());
String materializedTableDDL =
"CREATE MATERIALIZED TABLE users_shops"
@@ -457,9 +434,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
.containsKey("table.catalog-store.file.path")
.doesNotContainKey(WORKFLOW_SCHEDULER_TYPE.key())
.doesNotContainKey(RESOURCES_DOWNLOAD_DIR.key());
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -512,7 +486,7 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
data.add(Row.of(3L, 3L, 3L, "2024-01-02"));
createAndVerifyCreateMaterializedTableWithData(
- "my_materialized_table",
+ "users_shops",
data,
Collections.singletonMap("ds", "yyyy-MM-dd"),
RefreshMode.CONTINUOUS);
@@ -522,7 +496,7 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
long currentTime = System.currentTimeMillis();
String alterStatement =
- "ALTER MATERIALIZED TABLE my_materialized_table REFRESH
PARTITION (ds = '2024-01-02')";
+ "ALTER MATERIALIZED TABLE users_shops REFRESH PARTITION (ds =
'2024-01-02')";
OperationHandle alterHandle = executeStatement(alterStatement);
awaitOperationTermination(service, sessionHandle, alterHandle);
List<RowData> result = fetchAllResults(service, sessionHandle,
alterHandle);
@@ -535,7 +509,7 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
// 2. verify the new job overwrite the data
CommonTestUtils.waitUtil(
() ->
- fetchTableData(sessionHandle, "SELECT * FROM
my_materialized_table").size()
+ fetchTableData(sessionHandle, "SELECT * FROM
users_shops").size()
== data.size(),
Duration.ofMillis(timeout),
Duration.ofMillis(pause),
@@ -543,12 +517,9 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
assertThat(
fetchTableData(
sessionHandle,
- "SELECT * FROM my_materialized_table
where ds = '2024-01-02'")
+ "SELECT * FROM users_shops where ds =
'2024-01-02'")
.size())
.isEqualTo(1);
-
- // drop the materialized table
- dropMaterializedTable(getObjectIdentifier("my_materialized_table"));
}
@Test
@@ -628,9 +599,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
"Currently, refreshing materialized table only
supports referring to char, varchar and string type partition keys. All
specified partition keys in partition specs with unsupported types are:\n"
+ "\n"
+ "ds2");
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -736,9 +704,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
getJobRestoreSavepointPath(restClusterClient, resumeJobId);
assertThat(actualRestorePath).isNotEmpty();
assertThat(actualRestorePath.get()).isEqualTo(actualSavepointPath);
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -790,9 +755,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
"Savepoint directory is not configured, can't stop job
with savepoint.");
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -958,9 +920,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
fromJson((String)
jobDetail.getJobDataMap().get(WORKFLOW_INFO), WorkflowInfo.class);
assertThat(workflowInfo.getDynamicOptions())
.containsEntry("debezium-json.ignore-parse-errors", "true");
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -1017,8 +976,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
String.format(
"Materialized table %s refresh workflow has
been resumed.",
userShopsIdentifier));
-
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -1100,6 +1057,100 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
assertThat(oldTable.getDefinitionFreshness()).isEqualTo(newTable.getDefinitionFreshness());
}
+ @Test
+ void
testCreateOrAlterMaterializedTableColumnListReordersSchemaOnCreatePath() throws
Exception {
+ createBoundedValuesSource(List.of());
+
+ // users_shops does not exist yet, so CREATE OR ALTER falls through to
the create path and
+ // honors the DDL column list order, exactly like a plain CREATE.
+ String materializedTableDDL =
+ "CREATE OR ALTER MATERIALIZED TABLE users_shops (shop_id,
user_id, oca, order_id)"
+ + " WITH(\n"
+ + " 'format' = 'debezium-json'\n"
+ + " )\n"
+ + " AS SELECT \n"
+ + " user_id,\n"
+ + " shop_id,\n"
+ + " order_created_at AS oca,\n"
+ + " order_id"
+ + " FROM my_source";
+ OperationHandle handle = executeStatement(materializedTableDDL);
+ awaitOperationTermination(service, sessionHandle, handle);
+
+ ResolvedCatalogMaterializedTable table =
getTable(getObjectIdentifier("users_shops"));
+ assertThat(table.getResolvedSchema().getColumnNames())
+ .containsExactly("shop_id", "user_id", "oca", "order_id");
+ }
+
+ @Test
+ void testCreateOrAlterMaterializedTableColumnListIgnoredOnAlterPath()
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");
+
+ // users_shops already exists as a materialized table, so CREATE OR
ALTER takes the alter
+ // path. The DDL column list is treated as names only: it does NOT
reorder the schema.
+ String materializedTableDDL =
+ "CREATE OR ALTER MATERIALIZED TABLE users_shops (shop_id,
user_id, ds, order_cnt)"
+ + " PARTITIONED BY (ds)\n"
+ + " WITH(\n"
+ + " 'format' = 'debezium-json'\n"
+ + " )\n"
+ + " AS SELECT \n"
+ + " user_id,\n"
+ + " shop_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);
+ awaitOperationTermination(service, sessionHandle, handle);
+
+ ResolvedSchema newSchema =
getTable(userShopsIdentifier).getResolvedSchema();
+ assertThat(newSchema.getColumnNames())
+ .containsExactly("user_id", "shop_id", "ds", "order_cnt");
+ assertThat(newSchema).isEqualTo(oldSchema);
+ }
+
+ @Test
+ void testAlterMaterializedTableAsQueryRejectsReorder() throws Exception {
+ createAndVerifyCreateMaterializedTableWithData(
+ "users_shops", List.of(), Map.of(), RefreshMode.FULL);
+
+ ObjectIdentifier userShopsIdentifier =
getObjectIdentifier("users_shops");
+
assertThat(getTable(userShopsIdentifier).getResolvedSchema().getColumnNames())
+ .containsExactly("user_id", "shop_id", "ds", "order_cnt");
+
+ // ALTER ... AS re-derives the schema from the new query. Swapping the
first two
+ // projections asks to reorder existing columns. The query-evolution
rules reject this: an
+ // existing materialized table may only append columns at the end, not
reorder them.
+ String alterMaterializedTableAsQueryDDL =
+ "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(alterMaterializedTableAsQueryDDL);
+
+ assertThatThrownBy(() -> awaitOperationTermination(service,
sessionHandle, handle))
+ .hasStackTraceContaining("reordering columns are not
supported");
+
+ // The rejected alter leaves the original schema untouched.
+
assertThat(getTable(userShopsIdentifier).getResolvedSchema().getColumnNames())
+ .containsExactly("user_id", "shop_id", "ds", "order_cnt");
+ }
+
@Test
void testCreateMaterializedTableWithDistribution() {
String materializedTableDDL =
@@ -1337,9 +1388,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
getJobRestoreSavepointPath(
restClusterClient,
newContinuousRefreshHandler.getJobId());
assertThat(restorePath).isEmpty();
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -1412,9 +1460,6 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
ContinuousRefreshHandler newContinuousRefreshHandler =
getContinuousRefreshHandler(newTable);
assertThat(newContinuousRefreshHandler.getRestorePath()).isEmpty();
-
- // drop the materialized table
- dropMaterializedTable(userShopsIdentifier);
}
@Test
@@ -1618,12 +1663,12 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
data.add(Row.of(2L, 2L, 2L, "2024-01-02"));
createAndVerifyCreateMaterializedTableWithData(
- "my_materialized_table",
+ "users_shops",
data,
Collections.singletonMap("ds", "yyyy-MM-dd"),
RefreshMode.CONTINUOUS);
- ObjectIdentifier objectIdentifier =
getObjectIdentifier("my_materialized_table");
+ ObjectIdentifier objectIdentifier = getObjectIdentifier("users_shops");
// add more data to all data list
data.add(Row.of(3L, 3L, 3L, "2024-01-01"));
@@ -1655,7 +1700,7 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
assertThat(
fetchTableData(
sessionHandle,
- "SELECT * FROM my_materialized_table
where ds = '2024-01-02'")
+ "SELECT * FROM users_shops where ds =
'2024-01-02'")
.size())
.isEqualTo(getPartitionSize(data, "2024-01-02"));
@@ -1663,12 +1708,9 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
assertThat(
fetchTableData(
sessionHandle,
- "SELECT * FROM my_materialized_table
where ds = '2024-01-01'")
+ "SELECT * FROM users_shops where ds =
'2024-01-01'")
.size())
.isNotEqualTo(getPartitionSize(data, "2024-01-01"));
-
- // drop the materialized table
- dropMaterializedTable(objectIdentifier);
}
@Test
@@ -1679,13 +1721,9 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
// create materialized table without partition formatter
createAndVerifyCreateMaterializedTableWithData(
- "my_materialized_table_without_partition_options",
- data,
- Map.of(),
- RefreshMode.CONTINUOUS);
+ "users_shops", data, Map.of(), RefreshMode.CONTINUOUS);
- ObjectIdentifier tableWithoutFormatterIdentifier =
-
getObjectIdentifier("my_materialized_table_without_partition_options");
+ ObjectIdentifier tableWithoutFormatterIdentifier =
getObjectIdentifier("users_shops");
// add more data to all data list
data.add(Row.of(3L, 3L, 3L, "2024-01-01"));
@@ -1719,19 +1757,16 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
assertThat(
fetchTableData(
sessionHandle,
- "SELECT * FROM
my_materialized_table_without_partition_options where ds = '2024-01-01'")
+ "SELECT * FROM users_shops where ds =
'2024-01-01'")
.size())
.isEqualTo(getPartitionSize(data, "2024-01-01"));
assertThat(
fetchTableData(
sessionHandle,
- "SELECT * FROM
my_materialized_table_without_partition_options where ds = '2024-01-02'")
+ "SELECT * FROM users_shops where ds =
'2024-01-02'")
.size())
.isEqualTo(getPartitionSize(data, "2024-01-02"));
-
- // drop the materialized table
- dropMaterializedTable(tableWithoutFormatterIdentifier);
}
@Test
@@ -1740,12 +1775,12 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
// create materialized table with partition formatter
createAndVerifyCreateMaterializedTableWithData(
- "my_materialized_table",
+ "users_shops",
data,
Collections.singletonMap("ds", "yyyy-MM-dd"),
RefreshMode.FULL);
- ObjectIdentifier materializedTableIdentifier =
getObjectIdentifier("my_materialized_table");
+ ObjectIdentifier materializedTableIdentifier =
getObjectIdentifier("users_shops");
// add more data to all data list
data.add(Row.of(1L, 1L, 1L, "2024-01-01"));
@@ -1780,20 +1815,16 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
// data of partition '2024-01-01' is not changed
assertThat(
fetchTableData(
- sessionHandle,
- "SELECT * FROM my_materialized_table where ds
= '2024-01-02'"))
+ sessionHandle, "SELECT * FROM users_shops
where ds = '2024-01-02'"))
.size()
.isEqualTo(getPartitionSize(data, "2024-01-02"));
assertThat(
fetchTableData(
sessionHandle,
- "SELECT * FROM my_materialized_table
where ds = '2024-01-01'")
+ "SELECT * FROM users_shops where ds =
'2024-01-01'")
.size())
.isNotEqualTo(getPartitionSize(data, "2024-01-01"));
-
- // drop the materialized table
- dropMaterializedTable(materializedTableIdentifier);
}
@Test
@@ -1803,12 +1834,12 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
data.add(Row.of(2L, 2L, 2L, "2024-01-02"));
createAndVerifyCreateMaterializedTableWithData(
- "my_materialized_table",
+ "users_shops",
data,
Collections.singletonMap("ds", "yyyy-MM-dd"),
RefreshMode.CONTINUOUS);
- ObjectIdentifier objectIdentifier =
getObjectIdentifier("my_materialized_table");
+ ObjectIdentifier objectIdentifier = getObjectIdentifier("users_shops");
// refresh the materialized table with schedule time not specified
OperationHandle invalidRefreshTableHandle1 =
@@ -1853,9 +1884,86 @@ class MaterializedTableStatementITCase extends
AbstractMaterializedTableStatemen
String.format(
"Failed to parse a valid partition value for
the field 'ds' in materialized table %s using the scheduler time '20240103
00:00:00.000' based on the date format 'yyyy-MM-dd HH:mm:ss'.",
objectIdentifier.asSerializableString()));
+ }
- // drop the materialized table
- dropMaterializedTable(getObjectIdentifier("my_materialized_table"));
+ @Test
+ void testExplainCreateOrAlterMaterializedTableOnExistingTable() throws
Exception {
+ createAndVerifyCreateMaterializedTableWithData(
+ "users_shops", List.of(), Map.of(), RefreshMode.FULL);
+ ObjectIdentifier userShopsIdentifier =
getObjectIdentifier("users_shops");
+ ResolvedCatalogMaterializedTable oldTable =
getTable(userShopsIdentifier);
+
+ // CREATE OR ALTER on an existing table behaves like an
alter-with-query. Re-state the same
+ // definition so EXPLAIN plans the sink over the (unchanged) table.
+ String explainDDL =
+ "EXPLAIN CREATE OR ALTER MATERIALIZED TABLE users_shops"
+ + " PARTITIONED BY (ds)"
+ + " WITH ('format' = 'debezium-json')"
+ + " FRESHNESS = INTERVAL '30' SECOND"
+ + " REFRESH_MODE = FULL"
+ + " AS SELECT"
+ + " user_id,"
+ + " shop_id,"
+ + " ds,"
+ + " COUNT(order_id) AS order_cnt"
+ + " FROM ("
+ + " SELECT user_id, shop_id, order_created_at AS
ds, order_id FROM my_source"
+ + " ) AS tmp"
+ + " GROUP BY (user_id, shop_id, ds)";
+
+ OperationHandle explainHandle = executeStatement(explainDDL);
+ awaitOperationTermination(service, sessionHandle, explainHandle);
+ List<RowData> explainResults = fetchAllResults(service, sessionHandle,
explainHandle);
+ assertThat(explainResults).hasSize(1);
+ assertThat(explainResults.get(0).getString(0).toString())
+ .contains(
+ "== Abstract Syntax Tree ==",
+ "== Optimized Physical Plan ==",
+ "== Optimized Execution Plan ==",
+ "Sink(table=[",
+ "users_shops");
+
+ // EXPLAIN is side-effect-free: the existing materialized table is
unchanged.
+ ResolvedCatalogMaterializedTable newTable =
getTable(userShopsIdentifier);
+
assertThat(newTable.getResolvedSchema()).isEqualTo(oldTable.getResolvedSchema());
+
assertThat(newTable.getExpandedQuery()).isEqualTo(oldTable.getExpandedQuery());
+ }
+
+ @Test
+ void testExplainAlterMaterializedTableAsQuery() throws Exception {
+ createAndVerifyCreateMaterializedTableWithData(
+ "users_shops", List.of(), Map.of(), RefreshMode.FULL);
+ ObjectIdentifier userShopsIdentifier =
getObjectIdentifier("users_shops");
+ ResolvedCatalogMaterializedTable oldTable =
getTable(userShopsIdentifier);
+
+ String explainDDL =
+ "EXPLAIN ALTER MATERIALIZED TABLE users_shops"
+ + " AS SELECT"
+ + " user_id,"
+ + " shop_id,"
+ + " ds,"
+ + " COUNT(order_id) AS order_cnt"
+ + " FROM ("
+ + " SELECT user_id, shop_id, order_created_at AS
ds, order_id FROM my_source"
+ + " ) AS tmp"
+ + " GROUP BY (user_id, shop_id, ds)";
+
+ OperationHandle explainHandle = executeStatement(explainDDL);
+ awaitOperationTermination(service, sessionHandle, explainHandle);
+ List<RowData> explainResults = fetchAllResults(service, sessionHandle,
explainHandle);
+ assertThat(explainResults).hasSize(1);
+ assertThat(explainResults.get(0).getString(0).toString())
+ .contains(
+ "== Abstract Syntax Tree ==",
+ "== Optimized Physical Plan ==",
+ "== Optimized Execution Plan ==",
+ "Sink(table=[",
+ "users_shops");
+
+ // EXPLAIN is side-effect-free: the materialized table definition is
unchanged.
+ ResolvedCatalogMaterializedTable newTable =
getTable(userShopsIdentifier);
+
assertThat(newTable.getResolvedSchema()).isEqualTo(oldTable.getResolvedSchema());
+
assertThat(newTable.getExpandedQuery()).isEqualTo(oldTable.getExpandedQuery());
}
private void setupSavepointDir(Path temporaryPath) throws Exception {
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 aef621ddc7d..2b8250ea98e 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
@@ -3138,16 +3138,25 @@ SqlNode SqlRichExplain() :
stmt = RichSqlInsert()
|
stmt = SqlCreate()
+ |
+ stmt = SqlAlterMaterializedTable()
)
{
if ((stmt instanceof SqlCreate)
&& !(stmt instanceof SqlCreateTableAs)
- && !(stmt instanceof SqlReplaceTableAs)) {
+ && !(stmt instanceof SqlReplaceTableAs)
+ && !(stmt instanceof SqlCreateOrAlterMaterializedTable)) {
throw SqlUtil.newContextException(
getPos(),
ParserResource.RESOURCE.explainCreateOrReplaceStatementUnsupported());
}
+ if ((stmt instanceof SqlAlterMaterializedTable) && !(stmt instanceof
SqlAlterMaterializedTableAsQuery)) {
+ throw SqlUtil.newContextException(
+ getPos(),
+
ParserResource.RESOURCE.explainAlterMaterializedTableUnsupported());
+ }
+
return new SqlRichExplain(getPos(), stmt, explainDetails);
}
}
diff --git
a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/utils/ParserResource.java
b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/utils/ParserResource.java
index d459ca28dc8..7dade5ead96 100644
---
a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/utils/ParserResource.java
+++
b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/utils/ParserResource.java
@@ -53,6 +53,10 @@ public interface ParserResource {
"Unsupported CREATE OR REPLACE statement for EXPLAIN. The
statement must define a query using the AS clause (i.e. CTAS/RTAS statements).")
Resources.ExInst<ParseException>
explainCreateOrReplaceStatementUnsupported();
+ @Resources.BaseMessage(
+ "Unsupported ALTER MATERIALIZED TABLE statement for EXPLAIN. The
statement must define a query using the AS clause.")
+ Resources.ExInst<ParseException>
explainAlterMaterializedTableUnsupported();
+
@Resources.BaseMessage(
"Columns identifiers without types in the schema are supported on
CTAS/RTAS statements only.")
Resources.ExInst<ParseException> columnsIdentifiersUnsupported();
diff --git
a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
index 764bfce65d4..92f8e34c770 100644
---
a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
+++
b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
@@ -35,6 +35,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Execution;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.List;
import java.util.Locale;
@@ -3184,29 +3185,42 @@ class FlinkSqlParserImplTest extends SqlParserTest {
.fails("(?s).*Encountered \"\\,\" at line 1, column 32.\n.*");
}
- @Test
- void testExplainCreateTableNoSupported() {
- this.sql("EXPLAIN CREATE TABLE t (id int^)^")
+ @ParameterizedTest
+ @ValueSource(strings = {"CREATE", "CREATE OR REPLACE"})
+ void testExplainCreateTableNotSupported(final String operation) {
+ this.sql(String.format("EXPLAIN %s TABLE t (id int^)^", operation))
.fails(
"Unsupported CREATE OR REPLACE statement for
EXPLAIN\\. The statement must define a query using the AS clause \\(i\\.e\\.
CTAS/RTAS statements\\)\\.");
}
- @Test
- void testExplainCreateTableAsSelect() {
- this.sql("EXPLAIN CREATE TABLE t AS SELECT * FROM b")
- .ok("EXPLAIN CREATE TABLE `T`\nAS\nSELECT *\nFROM `B`");
- }
-
- @Test
- void testExplainCreateOrReplaceTableAsSelect() {
- this.sql("EXPLAIN CREATE OR REPLACE TABLE t AS SELECT * FROM b")
- .ok("EXPLAIN CREATE OR REPLACE TABLE `T`\nAS\nSELECT *\nFROM
`B`");
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "ALTER MATERIALIZED TABLE t ^SUSPEND^",
+ "ALTER MATERIALIZED TABLE t ^RESUME^",
+ "ALTER MATERIALIZED TABLE t ADD category_id STRING METADATA
^VIRTUAL^",
+ "ALTER MATERIALIZED TABLE t MODIFY measurement double
^METADATA^"
+ })
+ void testExplainAlterMaterializedTableWithoutAsQueryNotSupported(
+ final String alterMaterializedTable) {
+ this.sql(String.format("EXPLAIN %s", alterMaterializedTable))
+ .fails(
+ "Unsupported ALTER MATERIALIZED TABLE statement for
EXPLAIN\\. The statement must define a query using the AS clause.");
}
- @Test
- void testExplainReplaceTableAsSelect() {
- this.sql("EXPLAIN REPLACE TABLE t AS SELECT * FROM b")
- .ok("EXPLAIN REPLACE TABLE `T`\nAS\nSELECT *\nFROM `B`");
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "CREATE",
+ "CREATE OR REPLACE",
+ "REPLACE",
+ "CREATE MATERIALIZED",
+ "ALTER MATERIALIZED",
+ "CREATE OR ALTER MATERIALIZED",
+ })
+ void testExplain(final String operation) {
+ this.sql(String.format("EXPLAIN %s TABLE t AS SELECT * FROM b",
operation))
+ .ok(String.format("EXPLAIN %s TABLE `T`\nAS\nSELECT *\nFROM
`B`", operation));
}
@Test
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperationVisitor.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperationVisitor.java
index 006f042b893..2fc6d9931cf 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperationVisitor.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperationVisitor.java
@@ -19,6 +19,9 @@
package org.apache.flink.table.operations;
import org.apache.flink.annotation.Internal;
+import
org.apache.flink.table.operations.materializedtable.AlterMaterializedTableChangeOperation;
+import
org.apache.flink.table.operations.materializedtable.ConvertTableToMaterializedTableOperation;
+import
org.apache.flink.table.operations.materializedtable.CreateMaterializedTableOperation;
/**
* Class that implements visitor pattern. It allows type safe logic on top of
tree of {@link
@@ -39,4 +42,10 @@ public interface ModifyOperationVisitor<T> {
T visit(CreateTableASOperation ctas);
T visit(ReplaceTableAsOperation rtas);
+
+ T visit(CreateMaterializedTableOperation createMaterializedTableOperation);
+
+ T visit(AlterMaterializedTableChangeOperation
alterMaterializedTableChangeOperation);
+
+ T visit(ConvertTableToMaterializedTableOperation
convertTableToMaterializedTableOperation);
}
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
index 264fc752bfa..9da8302ca60 100644
---
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
@@ -23,6 +23,7 @@ import
org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.operations.QueryOperation;
import java.util.List;
import java.util.function.Function;
@@ -40,8 +41,9 @@ public class AlterMaterializedTableAsQueryOperation extends
AlterMaterializedTab
public AlterMaterializedTableAsQueryOperation(
ObjectIdentifier tableIdentifier,
Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangesForTable,
- ResolvedCatalogMaterializedTable oldTable) {
- super(tableIdentifier, tableChangesForTable, oldTable);
+ ResolvedCatalogMaterializedTable oldTable,
+ QueryOperation sinkModifyQuery) {
+ super(tableIdentifier, tableChangesForTable, oldTable,
sinkModifyQuery);
}
@Override
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 684518e99b0..87d588a920d 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
@@ -37,6 +37,9 @@ import
org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType;
import org.apache.flink.table.catalog.TableChange.ModifyRefreshHandler;
import org.apache.flink.table.catalog.TableChange.ModifyRefreshStatus;
import org.apache.flink.table.catalog.TableChange.ModifyStartMode;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.ModifyOperationVisitor;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.operations.ddl.AlterTableChangeOperation;
import java.util.ArrayList;
@@ -50,9 +53,11 @@ import java.util.stream.IntStream;
* Alter materialized table with new table definition and table changes
represents the modification.
*/
@Internal
-public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTableOperation {
+public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTableOperation
+ implements ModifyOperation {
private final Function<ResolvedCatalogMaterializedTable,
List<TableChange>> tableChangeForTable;
+ private final QueryOperation sinkModifyQuery;
private ResolvedCatalogMaterializedTable oldTable;
private MaterializedTableChangeHandler handler;
private CatalogMaterializedTable newTable;
@@ -62,9 +67,23 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
ObjectIdentifier tableIdentifier,
Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangeForTable,
ResolvedCatalogMaterializedTable oldTable) {
+ // Metadata-only changes (options, schema, distribution, ...) carry no
sink-modifying query.
+ this(tableIdentifier, tableChangeForTable, oldTable, null);
+ }
+
+ public AlterMaterializedTableChangeOperation(
+ ObjectIdentifier tableIdentifier,
+ Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangeForTable,
+ ResolvedCatalogMaterializedTable oldTable,
+ QueryOperation sinkModifyQuery) {
super(tableIdentifier);
this.tableChangeForTable = tableChangeForTable;
this.oldTable = oldTable;
+ this.sinkModifyQuery = sinkModifyQuery;
+ }
+
+ public QueryOperation getSinkModifyQuery() {
+ return sinkModifyQuery;
}
public List<TableChange> getTableChanges() {
@@ -76,7 +95,7 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
public AlterMaterializedTableChangeOperation copyAsTableChangeOperation() {
return new AlterMaterializedTableChangeOperation(
- tableIdentifier, tableChangeForTable, oldTable);
+ tableIdentifier, tableChangeForTable, oldTable,
sinkModifyQuery);
}
public CatalogMaterializedTable getNewTable() {
@@ -144,6 +163,16 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
"%s %s\n%s", getOperationName(),
tableIdentifier.asSummaryString(), changes);
}
+ @Override
+ public QueryOperation getChild() {
+ return this.sinkModifyQuery;
+ }
+
+ @Override
+ public <T> T accept(final ModifyOperationVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
/** Hook for subclasses to provide a different new-table builder. */
protected CatalogMaterializedTable computeNewTable() {
return
MaterializedTableChangeHandler.buildNewMaterializedTable(getHandlerWithChanges());
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/ConvertTableToMaterializedTableOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/ConvertTableToMaterializedTableOperation.java
index 8a83ea0f8e8..f00aef5e9f2 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/ConvertTableToMaterializedTableOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/ConvertTableToMaterializedTableOperation.java
@@ -26,8 +26,11 @@ import
org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.operations.ExecutableOperation;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.ModifyOperationVisitor;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
+import org.apache.flink.table.operations.QueryOperation;
import java.util.LinkedHashMap;
import java.util.List;
@@ -41,24 +44,27 @@ import java.util.stream.Collectors;
*/
@Internal
public class ConvertTableToMaterializedTableOperation
- implements MaterializedTableOperation, ExecutableOperation {
+ implements MaterializedTableOperation, ExecutableOperation,
ModifyOperation {
private final ObjectIdentifier tableIdentifier;
private final ResolvedCatalogTable originalTable;
private final ResolvedCatalogMaterializedTable materializedTable;
private final Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangesForTable;
+ private final QueryOperation sinkModifyQuery;
private List<TableChange> tableChanges;
public ConvertTableToMaterializedTableOperation(
ObjectIdentifier tableIdentifier,
ResolvedCatalogTable originalTable,
ResolvedCatalogMaterializedTable materializedTable,
- Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangesBuilder) {
+ Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangesBuilder,
+ QueryOperation sinkModifyQuery) {
this.tableIdentifier = tableIdentifier;
this.originalTable = originalTable;
this.materializedTable = materializedTable;
this.tableChangesForTable = tableChangesBuilder;
+ this.sinkModifyQuery = sinkModifyQuery;
}
@Override
@@ -81,6 +87,20 @@ public class ConvertTableToMaterializedTableOperation
return materializedTable;
}
+ public QueryOperation getSinkModifyQuery() {
+ return sinkModifyQuery;
+ }
+
+ @Override
+ public QueryOperation getChild() {
+ return sinkModifyQuery;
+ }
+
+ @Override
+ public <T> T accept(ModifyOperationVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
public List<TableChange> getTableChanges() {
if (tableChanges == null) {
tableChanges = tableChangesForTable.apply(materializedTable);
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/CreateMaterializedTableOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/CreateMaterializedTableOperation.java
index 9fb68dc71b0..ff425fe53aa 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/CreateMaterializedTableOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/CreateMaterializedTableOperation.java
@@ -23,26 +23,33 @@ import org.apache.flink.table.api.internal.TableResultImpl;
import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.ModifyOperationVisitor;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.operations.ddl.CreateOperation;
-import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
/** Operation to describe a CREATE MATERIALIZED TABLE statement. */
@Internal
public class CreateMaterializedTableOperation
- implements CreateOperation, MaterializedTableOperation {
+ implements CreateOperation, MaterializedTableOperation,
ModifyOperation {
private final ObjectIdentifier tableIdentifier;
private final ResolvedCatalogMaterializedTable materializedTable;
+ private final QueryOperation sinkModifyingQuery;
public CreateMaterializedTableOperation(
- ObjectIdentifier tableIdentifier, ResolvedCatalogMaterializedTable
materializedTable) {
+ ObjectIdentifier tableIdentifier,
+ ResolvedCatalogMaterializedTable materializedTable,
+ QueryOperation sinkModifyQuery) {
this.tableIdentifier = tableIdentifier;
this.materializedTable = materializedTable;
+ this.sinkModifyingQuery = sinkModifyQuery;
}
@Override
@@ -60,6 +67,10 @@ public class CreateMaterializedTableOperation
return materializedTable;
}
+ public QueryOperation getSinkModifyingQuery() {
+ return sinkModifyingQuery;
+ }
+
@Override
public String asSummaryString() {
Map<String, Object> params = new LinkedHashMap<>();
@@ -67,9 +78,16 @@ public class CreateMaterializedTableOperation
params.put("identifier", tableIdentifier);
return OperationUtils.formatWithChildren(
- "CREATE MATERIALIZED TABLE",
- params,
- Collections.emptyList(),
- Operation::asSummaryString);
+ "CREATE MATERIALIZED TABLE", params, List.of(),
Operation::asSummaryString);
+ }
+
+ @Override
+ public QueryOperation getChild() {
+ return this.sinkModifyingQuery;
+ }
+
+ @Override
+ public <T> T accept(final ModifyOperationVisitor<T> visitor) {
+ return visitor.visit(this);
}
}
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/FullAlterMaterializedTableOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/FullAlterMaterializedTableOperation.java
index ade473ddeac..b524e0c0cbf 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/FullAlterMaterializedTableOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/materializedtable/FullAlterMaterializedTableOperation.java
@@ -23,6 +23,7 @@ import
org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.operations.QueryOperation;
import java.util.List;
import java.util.function.Function;
@@ -42,8 +43,9 @@ public class FullAlterMaterializedTableOperation extends
AlterMaterializedTableC
final Function<ResolvedCatalogMaterializedTable,
List<TableChange>> tableChangeForTable,
final ResolvedCatalogMaterializedTable oldTable,
final Function<ResolvedCatalogMaterializedTable,
CatalogMaterializedTable>
- newTableBuilder) {
- super(tableIdentifier, tableChangeForTable, oldTable);
+ newTableBuilder,
+ final QueryOperation sinkModifyQuery) {
+ super(tableIdentifier, tableChangeForTable, oldTable, sinkModifyQuery);
this.newTableBuilder = newTableBuilder;
}
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
index 8a513278bf7..88073daca8b 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperationValidationTest.java
@@ -194,6 +194,6 @@ class AlterMaterializedTableAsQueryOperationValidationTest {
private static AlterMaterializedTableAsQueryOperation operation(
ResolvedCatalogMaterializedTable oldTable, List<TableChange>
changes) {
return new AlterMaterializedTableAsQueryOperation(
- ObjectIdentifier.of("cat", "db", "mt"), ignored -> changes,
oldTable);
+ ObjectIdentifier.of("cat", "db", "mt"), ignored -> changes,
oldTable, null);
}
}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java
index 860b6131da0..cb579c644ca 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java
@@ -168,9 +168,9 @@ public final class DynamicSinkUtils {
return convertSinkToRel(
relBuilder,
input,
- Collections.emptyMap(), // dynamicOptions
+ Map.of(), // dynamicOptions
contextResolvedTable,
- Collections.emptyMap(), // staticPartitions
+ Map.of(), // staticPartitions
null, // targetColumns
false,
tableSink,
@@ -192,9 +192,9 @@ public final class DynamicSinkUtils {
return convertSinkToRel(
relBuilder,
input,
- Collections.emptyMap(),
+ Map.of(),
externalModifyOperation.getContextResolvedTable(),
- Collections.emptyMap(),
+ Map.of(),
null, // targetColumns
false,
tableSink,
@@ -249,7 +249,7 @@ public final class DynamicSinkUtils {
return convertSinkToRel(
relBuilder,
input,
- Collections.emptyMap(),
+ Map.of(),
contextResolvedTable,
staticPartitions,
null,
@@ -258,6 +258,36 @@ public final class DynamicSinkUtils {
null); // conflictStrategy
}
+ /**
+ * Converts the {@code AS} query of a materialized table (CREATE / CREATE
OR ALTER / ALTER ...
+ * AS) into a {@link RelNode} that writes into the table, adding helper
projections if
+ * necessary. The caller resolves the target table to its {@link
ResolvedCatalogTable} form (the
+ * new definition for an alter, the created definition for a create).
+ */
+ public static RelNode convertMaterializedTableAsToRel(
+ FlinkRelBuilder relBuilder,
+ RelNode input,
+ Catalog catalog,
+ ObjectIdentifier identifier,
+ ResolvedCatalogTable resolvedTable,
+ Map<String, String> staticPartitions,
+ boolean isOverwrite,
+ DynamicTableSink sink) {
+ final ContextResolvedTable contextResolvedTable =
+ ContextResolvedTable.permanent(identifier, catalog,
resolvedTable);
+
+ return convertSinkToRel(
+ relBuilder,
+ input,
+ Map.of(),
+ contextResolvedTable,
+ staticPartitions,
+ null,
+ isOverwrite,
+ sink,
+ null);
+ }
+
private static RelNode convertSinkToRel(
FlinkRelBuilder relBuilder,
RelNode input,
@@ -1235,7 +1265,7 @@ public final class DynamicSinkUtils {
if (sink instanceof SupportsWritingMetadata) {
return ((SupportsWritingMetadata) sink).listWritableMetadata();
}
- return Collections.emptyMap();
+ return Map.of();
}
private static void validateAndApplyMetadata(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java
index eb29049dfb2..074309ab9f4 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java
@@ -32,6 +32,8 @@ import org.apache.flink.sql.parser.ddl.SqlUseDatabase;
import org.apache.flink.sql.parser.ddl.SqlUseModules;
import org.apache.flink.sql.parser.ddl.catalog.SqlDropCatalog;
import org.apache.flink.sql.parser.ddl.catalog.SqlUseCatalog;
+import
org.apache.flink.sql.parser.ddl.materializedtable.SqlAlterMaterializedTableAsQuery;
+import
org.apache.flink.sql.parser.ddl.materializedtable.SqlCreateMaterializedTable;
import org.apache.flink.sql.parser.ddl.table.SqlCreateTableAs;
import org.apache.flink.sql.parser.ddl.table.SqlDropTable;
import org.apache.flink.sql.parser.ddl.view.SqlDropView;
@@ -584,8 +586,10 @@ public class SqlNodeToOperationConversion {
}
} else if (sqlNode.getKind().belongsTo(SqlKind.QUERY)) {
operation = convertSqlQuery(sqlExplain.getStatement());
- } else if ((sqlNode instanceof SqlCreateTableAs)
- || (sqlNode instanceof SqlReplaceTableAs)) {
+ } else if (sqlNode instanceof SqlCreateTableAs
+ || sqlNode instanceof SqlReplaceTableAs
+ || sqlNode instanceof SqlCreateMaterializedTable
+ || sqlNode instanceof SqlAlterMaterializedTableAsQuery) {
operation =
convert(flinkPlanner, catalogManager, sqlNode)
.orElseThrow(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
index f2a6e2595a1..d2be3838a9c 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
@@ -101,7 +101,8 @@ public class SqlNodeConvertUtils {
return OptionalInt.empty();
}
- static PlannerQueryOperation toQueryOperation(SqlNode validated,
ConvertContext context) {
+ public static PlannerQueryOperation toQueryOperation(
+ SqlNode validated, ConvertContext context) {
// transform to a relational tree
RelRoot relational = context.toRelRoot(validated);
return new PlannerQueryOperation(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
index 3860a3f93c9..eb631523b98 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlAlterMaterializedTableAsQueryConverter.java
@@ -26,6 +26,7 @@ import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.operations.Operation;
import
org.apache.flink.table.operations.materializedtable.AlterMaterializedTableAsQueryOperation;
+import org.apache.flink.table.planner.calcite.FlinkPlannerImpl;
import org.apache.flink.table.planner.operations.PlannerQueryOperation;
import
org.apache.flink.table.planner.operations.converters.SqlNodeConvertUtils;
import org.apache.flink.table.planner.utils.MaterializedTableUtils;
@@ -36,6 +37,8 @@ import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
+import static
org.apache.flink.table.planner.operations.converters.SqlNodeConvertUtils.toQueryOperation;
+
/** A converter for {@link SqlAlterMaterializedTableAsQuery}. */
public class SqlAlterMaterializedTableAsQueryConverter
extends
AbstractAlterMaterializedTableConverter<SqlAlterMaterializedTableAsQuery> {
@@ -46,8 +49,13 @@ public class SqlAlterMaterializedTableAsQueryConverter
ResolvedCatalogMaterializedTable oldTable,
ConvertContext context) {
final ObjectIdentifier identifier =
resolveIdentifier(sqlAlterTableAsQuery, context);
+ final FlinkPlannerImpl flinkPlanner = context.getFlinkPlanner();
+ final SqlNode asQuerySqlNode = sqlAlterTableAsQuery.getAsQuery();
+ final SqlNode validatedAsQuery = flinkPlanner.validate(asQuerySqlNode);
+ final PlannerQueryOperation asQuery =
toQueryOperation(validatedAsQuery, context);
+
return new AlterMaterializedTableAsQueryOperation(
- identifier, gatherTableChanges(sqlAlterTableAsQuery, context),
oldTable);
+ identifier, gatherTableChanges(sqlAlterTableAsQuery, context),
oldTable, asQuery);
}
@Override
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 eff024de3d1..286071d3f98 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
@@ -45,6 +45,8 @@ import
org.apache.flink.table.operations.materializedtable.ConvertTableToMateria
import
org.apache.flink.table.operations.materializedtable.CreateMaterializedTableOperation;
import
org.apache.flink.table.operations.materializedtable.FullAlterMaterializedTableOperation;
import
org.apache.flink.table.operations.materializedtable.MaterializedTableChangeHandler;
+import org.apache.flink.table.planner.calcite.FlinkPlannerImpl;
+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;
@@ -59,6 +61,8 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import static
org.apache.flink.table.planner.operations.converters.SqlNodeConvertUtils.toQueryOperation;
+
/** A converter for {@link SqlCreateOrAlterMaterializedTable}. */
public class SqlCreateOrAlterMaterializedTableConverter
extends
AbstractCreateMaterializedTableConverter<SqlCreateOrAlterMaterializedTable> {
@@ -121,11 +125,17 @@ public class SqlCreateOrAlterMaterializedTableConverter
final ObjectIdentifier identifier) {
final SchemaResolver schemaResolver =
context.getCatalogManager().getSchemaResolver();
final MergeContext mergeContext =
getMergeContext(sqlCreateOrAlterTable, context);
+ final SqlNode asQuerySqlNode = sqlCreateOrAlterTable.getAsQuery();
+ final FlinkPlannerImpl flinkPlanner = context.getFlinkPlanner();
+ final SqlNode validatedAsQuery = flinkPlanner.validate(asQuerySqlNode);
+ final PlannerQueryOperation asQuery =
toQueryOperation(validatedAsQuery, context);
+
return new FullAlterMaterializedTableOperation(
identifier,
currentTable -> buildTableChanges(currentTable, mergeContext,
schemaResolver),
oldTable,
- currentTable -> buildNewTable(currentTable, mergeContext,
schemaResolver));
+ currentTable -> buildNewTable(currentTable, mergeContext,
schemaResolver),
+ asQuery);
}
private Operation handleConvert(
@@ -162,6 +172,19 @@ public class SqlCreateOrAlterMaterializedTableConverter
final ResolvedCatalogMaterializedTable resolvedNewMaterializedTable =
context.getCatalogManager().resolveCatalogMaterializedTable(newMaterializedTable);
+ final SqlNode asQuerySqlNode =
sqlCreateOrAlterMaterializedTable.getAsQuery();
+ final FlinkPlannerImpl flinkPlanner = context.getFlinkPlanner();
+ final SqlNode validatedAsQuery = flinkPlanner.validate(asQuerySqlNode);
+ final PlannerQueryOperation asQuery =
toQueryOperation(validatedAsQuery, context);
+ final PlannerQueryOperation sinkQuery =
+ new MergeTableAsUtil(context)
+ .maybeRewriteQuery(
+ context.getCatalogManager(),
+ flinkPlanner,
+ asQuery,
+ validatedAsQuery,
+ resolvedNewMaterializedTable);
+
return new ConvertTableToMaterializedTableOperation(
identifier,
oldBaseTable,
@@ -171,7 +194,8 @@ public class SqlCreateOrAlterMaterializedTableConverter
oldBaseTable,
resolvedCatalogMaterializedTable,
baseMergeContext.hasSchemaDefinition(),
- baseMergeContext.hasConstraintDefinition()));
+ baseMergeContext.hasConstraintDefinition()),
+ sinkQuery);
}
private List<TableChange> buildConversionTableChanges(
@@ -237,8 +261,20 @@ public class SqlCreateOrAlterMaterializedTableConverter
final ObjectIdentifier identifier) {
final ResolvedCatalogMaterializedTable resolvedTable =
getResolvedCatalogMaterializedTable(sqlCreateOrAlterTable,
context);
-
- return new CreateMaterializedTableOperation(identifier, resolvedTable);
+ final SqlNode asQuerySqlNode = sqlCreateOrAlterTable.getAsQuery();
+ final FlinkPlannerImpl flinkPlanner = context.getFlinkPlanner();
+ final SqlNode validatedAsQuery = flinkPlanner.validate(asQuerySqlNode);
+ final PlannerQueryOperation asQuery =
toQueryOperation(validatedAsQuery, context);
+ final PlannerQueryOperation sinkQuery =
+ new MergeTableAsUtil(context)
+ .maybeRewriteQuery(
+ context.getCatalogManager(),
+ flinkPlanner,
+ asQuery,
+ validatedAsQuery,
+ resolvedTable);
+
+ return new CreateMaterializedTableOperation(identifier, resolvedTable,
sinkQuery);
}
private List<TableChange> buildTableChanges(
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
index fe17d1f29d8..65c36d73151 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
@@ -35,6 +35,7 @@ import org.apache.flink.table.module.{Module, ModuleManager}
import org.apache.flink.table.operations._
import
org.apache.flink.table.operations.OutputConversionModifyOperation.UpdateMode
import org.apache.flink.table.operations.ddl.CreateTableOperation
+import
org.apache.flink.table.operations.materializedtable.{AlterMaterializedTableChangeOperation,
ConvertTableToMaterializedTableOperation, CreateMaterializedTableOperation}
import org.apache.flink.table.planner.JMap
import org.apache.flink.table.planner.calcite._
import org.apache.flink.table.planner.catalog.CatalogManagerCalciteSchema
@@ -51,9 +52,9 @@ import
org.apache.flink.table.planner.plan.nodes.physical.FlinkPhysicalRel
import org.apache.flink.table.planner.plan.optimize.Optimizer
import org.apache.flink.table.planner.sinks.DataStreamTableSink
import
org.apache.flink.table.planner.sinks.TableSinkUtils.{inferSinkPhysicalSchema,
validateLogicalPhysicalTypesCompatible, validateTableSink}
+import org.apache.flink.table.planner.utils.{JavaScalaConversionUtil,
TableConfigUtils}
import
org.apache.flink.table.planner.utils.InternalConfigOptions.{TABLE_QUERY_CURRENT_DATABASE,
TABLE_QUERY_START_EPOCH_TIME, TABLE_QUERY_START_LOCAL_TIME}
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil.{toJava,
toScala}
-import org.apache.flink.table.planner.utils.TableConfigUtils
import org.apache.flink.table.runtime.generated.CompileUtils
import org.apache.flink.table.types.utils.LegacyTypeInfoDataTypeConverter
@@ -285,8 +286,45 @@ abstract class PlannerBase(
convertCreateTableAsToRel(
rtasOperation.getCreateTableOperation,
rtasOperation.getChild,
- Collections.emptyMap(),
- false)
+ Map.empty[String, String].asJava,
+ isOverwrite = false)
+
+ case createMtOperation: CreateMaterializedTableOperation =>
+ convertMaterializedTableToRel(
+ createMtOperation.getTableIdentifier,
+ createMtOperation.getCatalogMaterializedTable.toResolvedCatalogTable,
+ createMtOperation.getChild,
+ Map.empty[String, String].asJava
+ )
+
+ case alterMtOperation: AlterMaterializedTableChangeOperation
+ if alterMtOperation.getSinkModifyQuery != null =>
+ val newTable =
+
catalogManager.resolveCatalogMaterializedTable(alterMtOperation.getNewTable)
+ // The alter reuses the existing storage, so keep the old table's
resolved connector
+ // options (e.g. managed `path`); new options override.
+ val sinkOptions =
+ new util.HashMap[String,
String](alterMtOperation.getOldTable.getOptions)
+ sinkOptions.putAll(newTable.getOptions)
+ convertMaterializedTableToRel(
+ alterMtOperation.getTableIdentifier,
+ newTable.toResolvedCatalogTable.copy(sinkOptions),
+ alterMtOperation.getChild,
+ Map.empty[String, String].asJava
+ )
+
+ case convertMtOperation: ConvertTableToMaterializedTableOperation
+ if convertMtOperation.getSinkModifyQuery != null =>
+ // The conversion reuses the original table's storage, so keep its
connector options.
+ val sinkOptions =
+ new util.HashMap[String,
String](convertMtOperation.getOriginalTable.getOptions)
+ sinkOptions.putAll(convertMtOperation.getMaterializedTable.getOptions)
+ convertMaterializedTableToRel(
+ convertMtOperation.getTableIdentifier,
+
convertMtOperation.getMaterializedTable.toResolvedCatalogTable.copy(sinkOptions),
+ convertMtOperation.getChild,
+ Map.empty[String, String].asJava
+ )
case stagedSink: StagedSinkModifyOperation =>
val input =
createRelBuilder.queryOperation(modifyOperation.getChild).build()
@@ -528,7 +566,7 @@ abstract class PlannerBase(
factory,
tableIdentifier,
resolvedCatalogTable,
- Collections.emptyMap(),
+ Map.empty[String, String].asJava,
getTableConfig,
getFlinkContext.getClassLoader,
isTemporary)
@@ -558,6 +596,28 @@ abstract class PlannerBase(
tableSink);
}
+ private def convertMaterializedTableToRel(
+ identifier: ObjectIdentifier,
+ resolvedTable: ResolvedCatalogTable,
+ queryOperation: QueryOperation,
+ staticPartitions: JMap[String, String]): RelNode = {
+ val input = createRelBuilder.queryOperation(queryOperation).build()
+ val catalog = toScala(catalogManager.getCatalog(identifier.getCatalogName))
+
+ val tableSink =
+ createDynamicTableSink(identifier, catalog, resolvedTable, isTemporary =
false)
+
+ DynamicSinkUtils.convertMaterializedTableAsToRel(
+ createRelBuilder,
+ input,
+ catalog.orNull,
+ identifier,
+ resolvedTable,
+ staticPartitions,
+ false,
+ tableSink)
+ }
+
protected def createSerdeContext: SerdeContext = {
val planner = createFlinkPlanner
new SerdeContext(
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 6e99aa29ed3..1d2cd769da8 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
@@ -43,6 +43,7 @@ import org.apache.flink.table.catalog.WatermarkSpec;
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.ExplainOperation;
import org.apache.flink.table.operations.Operation;
import
org.apache.flink.table.operations.materializedtable.AlterMaterializedTableAsQueryOperation;
import
org.apache.flink.table.operations.materializedtable.AlterMaterializedTableChangeOperation;
@@ -51,6 +52,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.operations.materializedtable.FullAlterMaterializedTableOperation;
import org.apache.flink.table.planner.utils.TableFunc0;
import org.junit.jupiter.api.BeforeEach;
@@ -757,6 +759,46 @@ class SqlMaterializedTableNodeToOperationConverterTest
assertThat(materializedTable.getOrigin()).isEqualTo(expected);
}
+ @Test
+ void testExplainCreateOrAlterMaterializedTable() {
+ final Operation operation =
+ parse("EXPLAIN CREATE OR ALTER MATERIALIZED TABLE myTable123
AS SELECT 123");
+
+ assertThat(operation).isInstanceOf(ExplainOperation.class);
+ assertThat(((ExplainOperation) operation).getChild())
+ .isInstanceOf(CreateMaterializedTableOperation.class);
+ }
+
+ @Test
+ void testExplainCreateMaterializedTable() {
+ final Operation operation =
+ parse("EXPLAIN CREATE MATERIALIZED TABLE myTable123 AS SELECT
123");
+
+ assertThat(operation).isInstanceOf(ExplainOperation.class);
+ assertThat(((ExplainOperation) operation).getChild())
+ .isInstanceOf(CreateMaterializedTableOperation.class);
+ }
+
+ @Test
+ void testExplainFullAlterMaterializedTable() {
+ // base_mtbl exists so the ALTER path is taken
+ final Operation operation =
+ parse("EXPLAIN CREATE OR ALTER MATERIALIZED TABLE base_mtbl AS
SELECT 123");
+ assertThat(operation).isInstanceOf(ExplainOperation.class);
+ assertThat(((ExplainOperation) operation).getChild())
+ .isInstanceOf(FullAlterMaterializedTableOperation.class);
+ }
+
+ @Test
+ void testExplainAlterMaterializedTable() {
+ final Operation operation =
+ parse("EXPLAIN ALTER MATERIALIZED TABLE base_mtbl AS SELECT
123");
+
+ assertThat(operation).isInstanceOf(ExplainOperation.class);
+ assertThat(((ExplainOperation) operation).getChild())
+ .isInstanceOf(AlterMaterializedTableChangeOperation.class);
+ }
+
private static Collection<TestSpec>
testDataForCreateAlterMaterializedTableFailedCase() {
final Collection<TestSpec> list = new ArrayList<>();
list.addAll(createWithInvalidSchema());
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java
new file mode 100644
index 00000000000..e3b35dd9f6b
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.plan.stream.sql;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.planner.utils.TableTestUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for EXPLAIN statements. */
+public class ExplainTest extends TableTestBase {
+
+ /**
+ * When the {@code PLAN_TEST_FORCE_OVERWRITE} environment variable is set
{@link #verifyExplain}
+ * overwrites the golden files.
+ */
+ private static final boolean REGENERATE_FILES =
+
"true".equalsIgnoreCase(System.getenv("PLAN_TEST_FORCE_OVERWRITE"));
+
+ private StreamTableTestUtil util;
+ private TestInfo testInfo;
+
+ @BeforeEach
+ void setup(TestInfo testInfo) {
+ this.testInfo = testInfo;
+ this.util = streamTestUtil(TableConfig.getDefault());
+ this.util
+ .getTableEnv()
+ .executeSql(
+ "CREATE TABLE MyTable (\n"
+ + " a INT,\n"
+ + " b BIGINT,\n"
+ + " c STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ")");
+ }
+
+ @Test
+ void testExplainCreateMaterializedTable() {
+ verifyExplain(
+ "CREATE MATERIALIZED TABLE MyMTTable\n"
+ + " WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable");
+ }
+
+ @Test
+ void testExplainCreateOrAlterMaterializedTable() {
+ verifyExplain(
+ "CREATE OR ALTER MATERIALIZED TABLE MyMTTable (\n"
+ + " `b`,\n"
+ + " `a`\n"
+ + " )\n"
+ + " WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " CAST(`a` AS BIGINT) AS `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable");
+ }
+
+ @Test
+ void testExplainAlterMaterializedTable() {
+ util.getTableEnv()
+ .executeSql(
+ "CREATE OR ALTER MATERIALIZED TABLE MyMTTable\n"
+ + " WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable");
+ verifyExplain(
+ "ALTER MATERIALIZED TABLE MyMTTable\n"
+ + "AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`,\n"
+ + " `c`\n"
+ + " FROM\n"
+ + " MyTable");
+ }
+
+ @Test
+ void testExplainFullAlterMaterializedTable() {
+ util.getTableEnv()
+ .executeSql(
+ "CREATE OR ALTER MATERIALIZED TABLE MyMTTable\n"
+ + " WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ") AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable");
+ verifyExplain(
+ "CREATE OR ALTER MATERIALIZED TABLE MyMTTable(\n"
+ + " `b`,\n"
+ + " `a`,\n"
+ + " `c`\n"
+ + " )\n"
+ + " WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ")\n"
+ + "AS\n"
+ + " SELECT\n"
+ + " CAST(`a` AS BIGINT) AS `a`,\n"
+ + " `b`,\n"
+ + " `c`\n"
+ + " FROM\n"
+ + " MyTable");
+ }
+
+ @Test
+ void testExplainConvertTableToMaterializedTable() {
+ final Configuration rootConfiguration = new Configuration();
+ rootConfiguration.set(
+
TableConfigOptions.MATERIALIZED_TABLE_CONVERSION_FROM_TABLE_ENABLED, true);
+ util.getTableEnv().getConfig().setRootConfiguration(rootConfiguration);
+ util.getTableEnv()
+ .executeSql(
+ "CREATE TABLE MyConvertTable (\n"
+ + " `a` INT,\n"
+ + " `b` BIGINT\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values'\n"
+ + ")");
+ verifyExplain(
+ "CREATE OR ALTER MATERIALIZED TABLE MyConvertTable\n"
+ + " AS\n"
+ + " SELECT\n"
+ + " `a`,\n"
+ + " `b`\n"
+ + " FROM\n"
+ + " MyTable");
+ }
+
+ private void verifyExplain(final String statement) {
+ final String actual = util.getTableEnv().explainSql(statement);
+ final String displayName = this.testInfo.getDisplayName();
+ final String fileName = displayName.substring(0, displayName.length()
- 2) + ".out";
+ if (REGENERATE_FILES) {
+ writeToResource(fileName, actual);
+ return;
+ }
+ final String expected = TableTestUtil.readFromResource("/explain/" +
fileName);
+ assertThat(TableTestUtil.replaceStageId(actual))
+ .isEqualTo(TableTestUtil.replaceStageId(expected));
+ }
+
+ private void writeToResource(final String fileName, final String content) {
+ try {
+ final Path testClassesRoot =
Paths.get(getClass().getResource("/").toURI());
+ final Path resourcesRoot =
+ Paths.get(
+ testClassesRoot
+ .toString()
+ .replace("target/test-classes",
"src/test/resources"));
+
Files.writeString(resourcesRoot.resolve("explain").resolve(fileName), content);
+ } catch (final Exception e) {
+ throw new RuntimeException("Failed to regenerate golden file " +
fileName, e);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainAlterMaterializedTable.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainAlterMaterializedTable.out
new file mode 100644
index 00000000000..488b651fe05
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainAlterMaterializedTable.out
@@ -0,0 +1,12 @@
+== Abstract Syntax Tree ==
+LogicalSink(table=[default_catalog.default_database.MyMTTable], fields=[a, b,
c])
++- LogicalProject(a=[$0], b=[$1], c=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+
+== Optimized Physical Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[a, b, c])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+
+== Optimized Execution Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[a, b, c])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainConvertTableToMaterializedTable.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainConvertTableToMaterializedTable.out
new file mode 100644
index 00000000000..dbce86bc35b
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainConvertTableToMaterializedTable.out
@@ -0,0 +1,12 @@
+== Abstract Syntax Tree ==
+LogicalSink(table=[default_catalog.default_database.MyConvertTable],
fields=[a, b])
++- LogicalProject(a=[$0], b=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+
+== Optimized Physical Plan ==
+Sink(table=[default_catalog.default_database.MyConvertTable], fields=[a, b])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
+
+== Optimized Execution Plan ==
+Sink(table=[default_catalog.default_database.MyConvertTable], fields=[a, b])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCreateMaterializedTable.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCreateMaterializedTable.out
new file mode 100644
index 00000000000..8440931c02e
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCreateMaterializedTable.out
@@ -0,0 +1,12 @@
+== Abstract Syntax Tree ==
+LogicalSink(table=[default_catalog.default_database.MyMTTable], fields=[a, b])
++- LogicalProject(a=[$0], b=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+
+== Optimized Physical Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[a, b])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
+
+== Optimized Execution Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[a, b])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[a, b], metadata=[]]], fields=[a, b])
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCreateOrAlterMaterializedTable.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCreateOrAlterMaterializedTable.out
new file mode 100644
index 00000000000..3bd4f82b621
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCreateOrAlterMaterializedTable.out
@@ -0,0 +1,14 @@
+== Abstract Syntax Tree ==
+LogicalSink(table=[default_catalog.default_database.MyMTTable], fields=[b, a])
++- LogicalProject(b=[$1], a=[CAST($0):BIGINT])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+
+== Optimized Physical Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[b, a])
++- Calc(select=[b, CAST(a AS BIGINT) AS a])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[b, a], metadata=[]]], fields=[b, a])
+
+== Optimized Execution Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[b, a])
++- Calc(select=[b, CAST(a AS BIGINT) AS a])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable,
project=[b, a], metadata=[]]], fields=[b, a])
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainFullAlterMaterializedTable.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainFullAlterMaterializedTable.out
new file mode 100644
index 00000000000..c5daa68fd09
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainFullAlterMaterializedTable.out
@@ -0,0 +1,14 @@
+== Abstract Syntax Tree ==
+LogicalSink(table=[default_catalog.default_database.MyMTTable], fields=[a, b,
c])
++- LogicalProject(a=[CAST($0):BIGINT], b=[$1], c=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+
+== Optimized Physical Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[a, b, c])
++- Calc(select=[CAST(a AS BIGINT) AS a, b, c])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+
+== Optimized Execution Plan ==
+Sink(table=[default_catalog.default_database.MyMTTable], fields=[a, b, c])
++- Calc(select=[CAST(a AS BIGINT) AS a, b, c])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
