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 ba41c1f950b [FLINK-40039][table] Fix `CTAS`/`RTAS` when used with
set-semantic PTFs
ba41c1f950b is described below
commit ba41c1f950bab2663cf0c323076bce7347d2fef8
Author: Ramin Gharib <[email protected]>
AuthorDate: Tue Jul 7 22:46:14 2026 +0200
[FLINK-40039][table] Fix `CTAS`/`RTAS` when used with set-semantic PTFs
---
.../MaterializedTableManager.java | 6 +-
.../AlterMaterializedTableAsQueryOperation.java | 4 +-
.../AlterMaterializedTableChangeOperation.java | 14 +--
.../CreateMaterializedTableOperation.java | 12 +--
.../operations/converters/MergeTableAsUtil.java | 106 +++++++++------------
.../AbstractCreateMaterializedTableConverter.java | 20 ++--
...SqlCreateOrAlterMaterializedTableConverter.java | 52 ++++------
.../table/SqlCreateTableAsConverter.java | 14 +--
.../table/SqlReplaceTableAsConverter.java | 6 +-
.../table/planner/delegation/PlannerBase.scala | 2 +-
...erializedTableNodeToOperationConverterTest.java | 16 ++++
.../planner/runtime/stream/sql/CTASITCase.java | 65 +++++++++++++
.../planner/runtime/stream/sql/RTASITCase.java | 23 +++++
...ExplainCtasWithColumnsInCreateAndQueryParts.out | 12 +--
.../planner/plan/stream/sql/TableSinkTest.scala | 60 +++++++-----
15 files changed, 249 insertions(+), 163 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 56cf110880a..cc01f96e7ec 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
@@ -955,7 +955,7 @@ public class MaterializedTableManager {
op.getTableIdentifier(),
oldTable -> op.getTableChanges(),
suspendMaterializedTable,
- op.getSinkModifyQuery());
+ op.getAsQueryOperation());
operationExecutor.callExecutableOperation(
handle, alterMaterializedTableChangeOperation);
@@ -1014,7 +1014,7 @@ public class MaterializedTableManager {
tableIdentifier,
oldTable -> tableChanges,
oldMaterializedTable,
- op.getSinkModifyQuery());
+ op.getAsQueryOperation());
operationExecutor.callExecutableOperation(
handle, alterMaterializedTableChangeOperation);
@@ -1036,7 +1036,7 @@ public class MaterializedTableManager {
op.getTableIdentifier(),
oldTable -> List.of(),
oldMaterializedTable,
- op.getSinkModifyQuery());
+ op.getAsQueryOperation());
}
private TableChange.ModifyRefreshHandler generateResetSavepointTableChange(
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 9da8302ca60..0dcbfd72926 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
@@ -42,8 +42,8 @@ public class AlterMaterializedTableAsQueryOperation extends
AlterMaterializedTab
ObjectIdentifier tableIdentifier,
Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangesForTable,
ResolvedCatalogMaterializedTable oldTable,
- QueryOperation sinkModifyQuery) {
- super(tableIdentifier, tableChangesForTable, oldTable,
sinkModifyQuery);
+ QueryOperation asQueryOperation) {
+ super(tableIdentifier, tableChangesForTable, oldTable,
asQueryOperation);
}
@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 87d588a920d..299c72d6cfe 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
@@ -57,7 +57,7 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
implements ModifyOperation {
private final Function<ResolvedCatalogMaterializedTable,
List<TableChange>> tableChangeForTable;
- private final QueryOperation sinkModifyQuery;
+ private final QueryOperation asQueryOperation;
private ResolvedCatalogMaterializedTable oldTable;
private MaterializedTableChangeHandler handler;
private CatalogMaterializedTable newTable;
@@ -75,15 +75,15 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
ObjectIdentifier tableIdentifier,
Function<ResolvedCatalogMaterializedTable, List<TableChange>>
tableChangeForTable,
ResolvedCatalogMaterializedTable oldTable,
- QueryOperation sinkModifyQuery) {
+ QueryOperation asQueryOperation) {
super(tableIdentifier);
this.tableChangeForTable = tableChangeForTable;
this.oldTable = oldTable;
- this.sinkModifyQuery = sinkModifyQuery;
+ this.asQueryOperation = asQueryOperation;
}
- public QueryOperation getSinkModifyQuery() {
- return sinkModifyQuery;
+ public QueryOperation getAsQueryOperation() {
+ return asQueryOperation;
}
public List<TableChange> getTableChanges() {
@@ -95,7 +95,7 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
public AlterMaterializedTableChangeOperation copyAsTableChangeOperation() {
return new AlterMaterializedTableChangeOperation(
- tableIdentifier, tableChangeForTable, oldTable,
sinkModifyQuery);
+ tableIdentifier, tableChangeForTable, oldTable,
asQueryOperation);
}
public CatalogMaterializedTable getNewTable() {
@@ -165,7 +165,7 @@ public class AlterMaterializedTableChangeOperation extends
AlterMaterializedTabl
@Override
public QueryOperation getChild() {
- return this.sinkModifyQuery;
+ return this.asQueryOperation;
}
@Override
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 ff425fe53aa..070e0f4948e 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
@@ -41,15 +41,15 @@ public class CreateMaterializedTableOperation
private final ObjectIdentifier tableIdentifier;
private final ResolvedCatalogMaterializedTable materializedTable;
- private final QueryOperation sinkModifyingQuery;
+ private final QueryOperation asQueryOperation;
public CreateMaterializedTableOperation(
ObjectIdentifier tableIdentifier,
ResolvedCatalogMaterializedTable materializedTable,
- QueryOperation sinkModifyQuery) {
+ QueryOperation asQueryOperation) {
this.tableIdentifier = tableIdentifier;
this.materializedTable = materializedTable;
- this.sinkModifyingQuery = sinkModifyQuery;
+ this.asQueryOperation = asQueryOperation;
}
@Override
@@ -67,8 +67,8 @@ public class CreateMaterializedTableOperation
return materializedTable;
}
- public QueryOperation getSinkModifyingQuery() {
- return sinkModifyingQuery;
+ public QueryOperation getAsQueryOperation() {
+ return asQueryOperation;
}
@Override
@@ -83,7 +83,7 @@ public class CreateMaterializedTableOperation
@Override
public QueryOperation getChild() {
- return this.sinkModifyingQuery;
+ return this.asQueryOperation;
}
@Override
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableAsUtil.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableAsUtil.java
index 2b26585d6f4..e41e0e9ed08 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableAsUtil.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableAsUtil.java
@@ -28,30 +28,28 @@ import
org.apache.flink.sql.parser.ddl.position.SqlTableColumnPosition;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Schema.UnresolvedColumn;
import org.apache.flink.table.api.Schema.UnresolvedPhysicalColumn;
-import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.planner.calcite.FlinkCalciteSqlValidator;
-import org.apache.flink.table.planner.calcite.FlinkPlannerImpl;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
-import org.apache.flink.table.planner.calcite.SqlRewriterUtils;
import org.apache.flink.table.planner.operations.PlannerQueryOperation;
-import org.apache.flink.table.planner.operations.SqlNodeToOperationConversion;
import
org.apache.flink.table.planner.operations.converters.SqlNodeConverter.ConvertContext;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.RelFactories;
import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlIdentifier;
-import org.apache.calcite.sql.SqlLiteral;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlNodeList;
-import org.apache.calcite.sql.parser.SqlParserPos;
import org.apache.calcite.sql.validate.SqlValidator;
+import org.apache.calcite.tools.RelBuilder;
import javax.annotation.Nullable;
@@ -88,80 +86,70 @@ public class MergeTableAsUtil {
}
/**
- * Rewrites the query operation to include only the fields that may be
persisted in the sink.
+ * Reshapes the query so its output columns line up with the sink's
persistable columns:
+ * reordering them and filling sink columns the query does not produce
with {@code NULL}.
+ * Returns the query unchanged when it already matches the sink 1:1. A
sink column the query
+ * does not produce that is declared {@code NOT NULL} raises a {@link
ValidationException}.
*/
public PlannerQueryOperation maybeRewriteQuery(
- CatalogManager catalogManager,
- FlinkPlannerImpl flinkPlanner,
PlannerQueryOperation origQueryOperation,
SqlNode origQueryNode,
ResolvedCatalogBaseTable<?> sinkTable) {
- FlinkCalciteSqlValidator sqlValidator =
flinkPlanner.getOrCreateSqlValidator();
- SqlRewriterUtils rewriterUtils = new SqlRewriterUtils(sqlValidator);
- FlinkTypeFactory typeFactory = (FlinkTypeFactory)
sqlValidator.getTypeFactory();
-
- // Only fields that may be persisted will be included in the select
query
- RowType sinkRowType =
- ((RowType)
sinkTable.getResolvedSchema().toSinkRowDataType().getLogicalType());
-
- Map<String, Integer> sourceFields =
- IntStream.range(0,
origQueryOperation.getResolvedSchema().getColumnNames().size())
+ final RelNode queryRelNode = origQueryOperation.getCalciteTree();
+ final RelOptCluster cluster = queryRelNode.getCluster();
+ final RexBuilder rexBuilder = cluster.getRexBuilder();
+ final FlinkTypeFactory typeFactory = (FlinkTypeFactory)
cluster.getTypeFactory();
+
+ // Only fields that may be persisted are included in the sink.
+ final RowType sinkRowType =
+ (RowType)
sinkTable.getResolvedSchema().toSinkRowDataType().getLogicalType();
+
+ final List<String> sourceColumns =
origQueryOperation.getResolvedSchema().getColumnNames();
+ final Map<String, Integer> sourceFields =
+ IntStream.range(0, sourceColumns.size())
.boxed()
- .collect(
- Collectors.toMap(
-
origQueryOperation.getResolvedSchema().getColumnNames()
- ::get,
- Function.identity()));
+ .collect(Collectors.toMap(sourceColumns::get,
Function.identity()));
- // assignedFields contains the new sink fields that are not present in
the source
- // and that will be included in the select query
- LinkedHashMap<Integer, SqlNode> assignedFields = new LinkedHashMap<>();
-
- // targetPositions contains the positions of the source fields that
will be
- // included in the select query
- List<Object> targetPositions = new ArrayList<>();
+ final List<RexNode> projects = new ArrayList<>();
+ final List<String> fieldNames = new ArrayList<>();
+ // The projection is a no-op when the query already produces the sink
columns 1:1 in order.
+ boolean rewriteNeeded = sinkRowType.getFieldCount() !=
sourceColumns.size();
+ // The loop cannot stop once a rewrite is detected: the projection
must cover every sink
+ // field, and every missing NOT NULL column must still be validated.
int pos = -1;
for (RowType.RowField targetField : sinkRowType.getFields()) {
pos++;
+ fieldNames.add(targetField.getName());
- if (!sourceFields.containsKey(targetField.getName())) {
+ final Integer sourcePos = sourceFields.get(targetField.getName());
+ if (sourcePos == null) {
if (!targetField.getType().isNullable()) {
throw new ValidationException(
"Column '"
+ targetField.getName()
+ "' has no default value and does not
allow NULLs.");
}
-
- assignedFields.put(
- pos,
- validator.maybeCast(
- SqlLiteral.createNull(SqlParserPos.ZERO),
- typeFactory.createUnknownType(),
+ projects.add(
+ rexBuilder.makeNullLiteral(
typeFactory.createFieldTypeFromLogicalType(targetField.getType())));
+ rewriteNeeded = true;
} else {
- targetPositions.add(sourceFields.get(targetField.getName()));
+ projects.add(rexBuilder.makeInputRef(queryRelNode, sourcePos));
+ if (sourcePos != pos) {
+ rewriteNeeded = true;
+ }
}
}
- // rewrite query
- SqlCall newSelect =
- rewriterUtils.rewriteCall(
- rewriterUtils,
- sqlValidator,
- (SqlCall) origQueryNode,
- typeFactory.buildRelNodeRowType(sinkRowType),
- assignedFields,
- targetPositions,
- () -> "Unsupported node type " +
origQueryNode.getKind());
-
- return (PlannerQueryOperation)
- SqlNodeToOperationConversion.convert(flinkPlanner,
catalogManager, newSelect)
- .orElseThrow(
- () ->
- new TableException(
- "Unsupported node type "
- +
newSelect.getClass().getSimpleName()));
+ if (!rewriteNeeded) {
+ return origQueryOperation;
+ }
+
+ final RelBuilder relBuilder =
RelFactories.LOGICAL_BUILDER.create(cluster, null);
+ final RelNode projected =
+ relBuilder.push(queryRelNode).project(projects, fieldNames,
true).build();
+ return new PlannerQueryOperation(projected, () ->
escapeExpression.apply(origQueryNode));
}
/**
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/AbstractCreateMaterializedTableConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/AbstractCreateMaterializedTableConverter.java
index 3ed73870835..483f062b180 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/AbstractCreateMaterializedTableConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/AbstractCreateMaterializedTableConverter.java
@@ -80,6 +80,8 @@ public abstract class
AbstractCreateMaterializedTableConverter<T extends SqlCrea
ResolvedSchema getMergedQuerySchema();
+ PlannerQueryOperation getAsQueryOperation();
+
RefreshMode getMergedRefreshMode();
LogicalRefreshMode getMergedLogicalRefreshMode();
@@ -121,16 +123,13 @@ public abstract class
AbstractCreateMaterializedTableConverter<T extends SqlCrea
.get(MaterializedTableConfigOptions.MATERIALIZED_TABLE_DEFAULT_START_MODE));
}
- protected final ResolvedSchema getQueryResolvedSchema(
+ protected final PlannerQueryOperation getAsQueryOperation(
T sqlCreateMaterializedTable, ConvertContext context) {
- SqlNode selectQuery = sqlCreateMaterializedTable.getAsQuery();
- SqlNode validateQuery =
context.getSqlValidator().validate(selectQuery);
-
- PlannerQueryOperation queryOperation =
- new PlannerQueryOperation(
- context.toRelRoot(validateQuery).project(),
- () -> context.toQuotedSqlString(validateQuery));
- return queryOperation.getResolvedSchema();
+ final SqlNode selectQuery = sqlCreateMaterializedTable.getAsQuery();
+ final SqlNode validateQuery =
context.getSqlValidator().validate(selectQuery);
+ return new PlannerQueryOperation(
+ context.toRelRoot(validateQuery).project(),
+ () -> context.toQuotedSqlString(validateQuery));
}
protected final LogicalRefreshMode getDerivedLogicalRefreshMode(T
sqlCreateMaterializedTable) {
@@ -167,8 +166,7 @@ public abstract class
AbstractCreateMaterializedTableConverter<T extends SqlCrea
}
protected final ResolvedCatalogMaterializedTable
getResolvedCatalogMaterializedTable(
- T sqlCreateMaterializedTable, ConvertContext context) {
- final MergeContext mergeContext =
getMergeContext(sqlCreateMaterializedTable, context);
+ MergeContext mergeContext, T sqlCreateMaterializedTable,
ConvertContext context) {
final List<String> partitionKeys =
mergeContext.getMergedPartitionKeys();
final Schema schema = mergeContext.getMergedSchema();
final ResolvedSchema querySchema = mergeContext.getMergedQuerySchema();
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 286071d3f98..bb757cd91a4 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,7 +45,6 @@ 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;
@@ -61,8 +60,6 @@ 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> {
@@ -125,10 +122,7 @@ 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);
+ final PlannerQueryOperation asQuery =
mergeContext.getAsQueryOperation();
return new FullAlterMaterializedTableOperation(
identifier,
@@ -172,17 +166,11 @@ 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 =
+ final PlannerQueryOperation asQueryOperation =
new MergeTableAsUtil(context)
.maybeRewriteQuery(
- context.getCatalogManager(),
- flinkPlanner,
- asQuery,
- validatedAsQuery,
+ baseMergeContext.getAsQueryOperation(),
+ sqlCreateOrAlterMaterializedTable.getAsQuery(),
resolvedNewMaterializedTable);
return new ConvertTableToMaterializedTableOperation(
@@ -195,7 +183,7 @@ public class SqlCreateOrAlterMaterializedTableConverter
resolvedCatalogMaterializedTable,
baseMergeContext.hasSchemaDefinition(),
baseMergeContext.hasConstraintDefinition()),
- sinkQuery);
+ asQueryOperation);
}
private List<TableChange> buildConversionTableChanges(
@@ -259,22 +247,17 @@ public class SqlCreateOrAlterMaterializedTableConverter
final SqlCreateOrAlterMaterializedTable sqlCreateOrAlterTable,
final ConvertContext context,
final ObjectIdentifier identifier) {
+ final MergeContext mergeContext =
getMergeContext(sqlCreateOrAlterTable, context);
final ResolvedCatalogMaterializedTable resolvedTable =
- getResolvedCatalogMaterializedTable(sqlCreateOrAlterTable,
context);
- 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 =
+ getResolvedCatalogMaterializedTable(mergeContext,
sqlCreateOrAlterTable, context);
+ final PlannerQueryOperation asQueryOperation =
new MergeTableAsUtil(context)
.maybeRewriteQuery(
- context.getCatalogManager(),
- flinkPlanner,
- asQuery,
- validatedAsQuery,
+ mergeContext.getAsQueryOperation(),
+ sqlCreateOrAlterTable.getAsQuery(),
resolvedTable);
- return new CreateMaterializedTableOperation(identifier, resolvedTable,
sinkQuery);
+ return new CreateMaterializedTableOperation(identifier, resolvedTable,
asQueryOperation);
}
private List<TableChange> buildTableChanges(
@@ -438,10 +421,12 @@ public class SqlCreateOrAlterMaterializedTableConverter
SqlCreateOrAlterMaterializedTableConverter.this.getDerivedOriginalQuery(
sqlCreateMaterializedTable, context);
- private final ResolvedSchema querySchema =
-
SqlCreateOrAlterMaterializedTableConverter.this.getQueryResolvedSchema(
+ private final PlannerQueryOperation asQueryOperation =
+
SqlCreateOrAlterMaterializedTableConverter.this.getAsQueryOperation(
sqlCreateMaterializedTable, context);
+ private final ResolvedSchema querySchema =
asQueryOperation.getResolvedSchema();
+
@Override
public boolean hasSchemaDefinition() {
final SqlNodeList sqlNodeList =
sqlCreateMaterializedTable.getColumnList();
@@ -506,7 +491,12 @@ public class SqlCreateOrAlterMaterializedTableConverter
@Override
public ResolvedSchema getMergedQuerySchema() {
- return this.querySchema;
+ return asQueryOperation.getResolvedSchema();
+ }
+
+ @Override
+ public PlannerQueryOperation getAsQueryOperation() {
+ return asQueryOperation;
}
@Override
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlCreateTableAsConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlCreateTableAsConverter.java
index 8fa2702a329..b0761ebbfdd 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlCreateTableAsConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlCreateTableAsConverter.java
@@ -47,18 +47,17 @@ public class SqlCreateTableAsConverter extends
AbstractCreateTableConverter<SqlC
public Operation convertSqlNode(SqlCreateTableAs sqlCreateTableAs,
ConvertContext context) {
final FlinkPlannerImpl flinkPlanner = context.getFlinkPlanner();
final CatalogManager catalogManager = context.getCatalogManager();
- SqlNode asQuerySqlNode = sqlCreateTableAs.getAsQuery();
- SqlNode validatedAsQuery = flinkPlanner.validate(asQuerySqlNode);
+ final SqlNode asQuerySqlNode = sqlCreateTableAs.getAsQuery();
PlannerQueryOperation query =
(PlannerQueryOperation)
SqlNodeToOperationConversion.convert(
- flinkPlanner, catalogManager,
validatedAsQuery)
+ flinkPlanner, catalogManager,
asQuerySqlNode)
.orElseThrow(
() ->
new TableException(
"CTAS unsupported node
type "
- +
validatedAsQuery
+ +
asQuerySqlNode
.getClass()
.getSimpleName()));
ResolvedCatalogTable tableWithResolvedSchema =
@@ -67,12 +66,7 @@ public class SqlCreateTableAsConverter extends
AbstractCreateTableConverter<SqlC
// If needed, rewrite the query to include the new sink fields in the
select list
query =
new MergeTableAsUtil(context)
- .maybeRewriteQuery(
- catalogManager,
- flinkPlanner,
- query,
- validatedAsQuery,
- tableWithResolvedSchema);
+ .maybeRewriteQuery(query, asQuerySqlNode,
tableWithResolvedSchema);
ObjectIdentifier identifier = getIdentifier(sqlCreateTableAs, context);
CreateTableOperation createTableOperation =
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlReplaceTableAsConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlReplaceTableAsConverter.java
index 438575d8f94..34b9c10a84e 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlReplaceTableAsConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlReplaceTableAsConverter.java
@@ -69,11 +69,7 @@ public class SqlReplaceTableAsConverter extends
AbstractCreateTableConverter<Sql
query =
new MergeTableAsUtil(context)
.maybeRewriteQuery(
- context.getCatalogManager(),
- flinkPlanner,
- query,
- sqlReplaceTableAs.getAsQuery(),
- tableWithResolvedSchema);
+ query, sqlReplaceTableAs.getAsQuery(),
tableWithResolvedSchema);
ObjectIdentifier identifier = getIdentifier(sqlReplaceTableAs,
context);
CreateTableOperation createTableOperation =
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 65c36d73151..e4f76fd1990 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
@@ -298,7 +298,7 @@ abstract class PlannerBase(
)
case alterMtOperation: AlterMaterializedTableChangeOperation
- if alterMtOperation.getSinkModifyQuery != null =>
+ if alterMtOperation.getAsQueryOperation != null =>
val newTable =
catalogManager.resolveCatalogMaterializedTable(alterMtOperation.getNewTable)
// The alter reuses the existing storage, so keep the old table's
resolved connector
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 1d2cd769da8..3c4aeac8343 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
@@ -53,6 +53,7 @@ import
org.apache.flink.table.operations.materializedtable.AlterMaterializedTabl
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.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.SetSemanticTableFunction;
import org.apache.flink.table.planner.utils.TableFunc0;
import org.junit.jupiter.api.BeforeEach;
@@ -197,6 +198,21 @@ class SqlMaterializedTableNodeToOperationConverterTest
sqlWithNonPersistedLast, "base_mtbl_with_non_persisted_last");
}
+ @Test
+ void testCreateMaterializedTableAsSelectWithSetSemanticTablePtf() {
+ functionCatalog.registerTemporarySystemFunction("f", new
SetSemanticTableFunction(), false);
+ final String sql =
+ "CREATE MATERIALIZED TABLE mt_ptf\n"
+ + "WITH (\n"
+ + " 'connector' = 'filesystem',\n"
+ + " 'format' = 'json'\n"
+ + ")\n"
+ + "AS SELECT * FROM f(r => TABLE t1 PARTITION BY a, i
=> 1)";
+
+ final Operation operation = parse(sql);
+
assertThat(operation).isInstanceOf(CreateMaterializedTableOperation.class);
+ }
+
@Test
void testCreateMaterializedTable() {
final String sql =
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/CTASITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/CTASITCase.java
new file mode 100644
index 00000000000..e6f52b02649
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/CTASITCase.java
@@ -0,0 +1,65 @@
+/*
+ * 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.runtime.stream.sql;
+
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.SetSemanticTableFunction;
+import org.apache.flink.table.planner.runtime.utils.StreamingTestBase;
+import org.apache.flink.table.planner.runtime.utils.TestData;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT Case for CREATE TABLE AS SELECT statement over a {@link
ProcessTableFunction}. */
+class CTASITCase extends StreamingTestBase {
+
+ @BeforeEach
+ @Override
+ public void before() throws Exception {
+ super.before();
+ final String dataId =
TestValuesTableFactory.registerData(TestData.smallData3());
+ tEnv().executeSql(
+ String.format(
+ "CREATE TABLE source(a int, b bigint, c
string)"
+ + " WITH ('connector' = 'values',
'bounded' = 'true', 'data-id' = '%s')",
+ dataId));
+ tEnv().createTemporarySystemFunction("f",
SetSemanticTableFunction.class);
+ }
+
+ @Test
+ void testCreateTableAsSelectOverSetSemanticPtfWithReorderedColumns()
throws Exception {
+ tEnv().executeSql(
+ "CREATE TABLE sink (`out`, `a`) WITH ('connector' =
'values') AS "
+ + "SELECT * FROM f(r => TABLE source PARTITION
BY a, i => 1)")
+ .await();
+
+ final ResolvedSchema schema = tEnv().from("sink").getResolvedSchema();
+ assertThat(schema.getColumnNames()).containsExactly("out", "a");
+
+ assertThat(TestValuesTableFactory.getResultsAsStrings("sink"))
+ .containsExactlyInAnyOrder(
+ "+I[{+I[1, 1, Hi], 1}, 1]",
+ "+I[{+I[2, 2, Hello], 1}, 2]",
+ "+I[{+I[3, 2, Hello world], 1}, 3]");
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/RTASITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/RTASITCase.java
index c1a943d9f74..f416ab30d05 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/RTASITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/RTASITCase.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.SetSemanticTableFunction;
import org.apache.flink.table.planner.runtime.utils.StreamingTestBase;
import org.apache.flink.table.planner.runtime.utils.TestData;
import org.apache.flink.table.types.AbstractDataType;
@@ -253,6 +254,28 @@ class RTASITCase extends StreamingTestBase {
verifyCatalogTable(expectCatalogTable,
getCatalogTable("not_exist_target"));
}
+ @Test
+ void testReplaceTableAsSelectOverSetSemanticPtf() throws Exception {
+ tEnv().createTemporarySystemFunction("f",
SetSemanticTableFunction.class);
+ tEnv().executeSql(
+ "REPLACE TABLE target WITH ('connector' = 'values',
'bounded' = 'true') AS "
+ + "SELECT * FROM f(r => TABLE source PARTITION
BY a, i => 1)")
+ .await();
+
+ assertThat(TestValuesTableFactory.getResultsAsStrings("target"))
+ .containsExactlyInAnyOrder(
+ "+I[1, {+I[1, 1, Hi], 1}]",
+ "+I[2, {+I[2, 2, Hello], 1}]",
+ "+I[3, {+I[3, 2, Hello world], 1}]");
+
+ // The partition key `a` is passed through and the PTF appends its
`out` column.
+ CatalogTable expectCatalogTable =
+ getExpectCatalogTable(
+ new String[] {"a", "out"},
+ new AbstractDataType[] {DataTypes.INT(),
DataTypes.STRING()});
+ verifyCatalogTable(expectCatalogTable, getCatalogTable("target"));
+ }
+
private CatalogTable getExpectCatalogTable(
String[] cols, AbstractDataType<?>[] fieldDataTypes) {
return getExpectCatalogTable(cols, fieldDataTypes,
getDefaultTargetTableOptions());
diff --git
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
index 811d6694c59..eb7db8e3862 100644
---
a/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
+++
b/flink-table/flink-table-planner/src/test/resources/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out
@@ -1,14 +1,14 @@
== Abstract Syntax Tree ==
-LogicalSink(table=[default_catalog.default_database.MyCtasTable],
fields=[EXPR$0, a, b])
-+- LogicalProject(EXPR$0=[null:INTEGER], a=[$0], b=[$1])
+LogicalSink(table=[default_catalog.default_database.MyCtasTable],
fields=[votes, a, b, metadata_col])
++- LogicalProject(votes=[null:INTEGER], a=[$0], b=[$1],
metadata_col=[null:BIGINT])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
== Optimized Physical Plan ==
-Sink(table=[default_catalog.default_database.MyCtasTable], fields=[EXPR$0, a,
b])
-+- Calc(select=[null:INTEGER AS EXPR$0, a, b])
+Sink(table=[default_catalog.default_database.MyCtasTable], fields=[votes, a,
b, metadata_col])
++- Calc(select=[null:INTEGER AS votes, a, b, null:BIGINT AS metadata_col])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
== Optimized Execution Plan ==
-Sink(table=[default_catalog.default_database.MyCtasTable], fields=[EXPR$0, a,
b])
-+- Calc(select=[null:INTEGER AS EXPR$0, a, b])
+Sink(table=[default_catalog.default_database.MyCtasTable], fields=[votes, a,
b, metadata_col])
++- Calc(select=[null:INTEGER AS votes, a, b, null:BIGINT AS metadata_col])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
index ef9776caedc..5201b1f20bc 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
@@ -889,17 +889,25 @@ class TableSinkTest extends TableTestBase {
@Test
def testExplainCreateTableAsSelectWithColumnsInCreateAndQueryParts(): Unit =
{
val actual =
- util.tableEnv.explainSql("""
- |CREATE TABLE MyCtasTable(`votes` INT,
`votes_2x` AS `b` * 2)
- | WITH (
- | 'connector' = 'values'
- |) AS
- | SELECT
- | `a`,
- | `b`
- | FROM
- | MyTable
- |""".stripMargin)
+ util.tableEnv.explainSql(
+ """
+ |CREATE TABLE MyCtasTable(
+ | `votes` INT,
+ | `votes_2x` AS `b` * 2,
+ | `metadata_col` BIGINT METADATA,
+ | `virtual_col` STRING METADATA VIRTUAL
+ |)
+ | WITH (
+ | 'connector' = 'values',
+ | 'readable-metadata' = 'metadata_col:BIGINT, virtual_col:STRING',
+ | 'writable-metadata' = 'metadata_col:BIGINT'
+ |) AS
+ | SELECT
+ | `a`,
+ | `b`
+ | FROM
+ | MyTable
+ |""".stripMargin)
val expected =
TableTestUtil.readFromResource("/explain/testExplainCtasWithColumnsInCreateAndQueryParts.out")
@@ -910,17 +918,25 @@ class TableSinkTest extends TableTestBase {
@Test
def testExplainReplaceTableAsSelectWithColumnsInCreateAndQueryParts(): Unit
= {
val actual =
- util.tableEnv.explainSql("""
- |REPLACE TABLE MyCtasTable(`votes` INT,
`votes_2x` AS `b` * 2)
- | WITH (
- | 'connector' = 'values'
- |) AS
- | SELECT
- | `a`,
- | `b`
- | FROM
- | MyTable
- |""".stripMargin)
+ util.tableEnv.explainSql(
+ """
+ |REPLACE TABLE MyCtasTable(
+ | `votes` INT,
+ | `votes_2x` AS `b` * 2,
+ | `metadata_col` BIGINT METADATA,
+ | `virtual_col` STRING METADATA VIRTUAL
+ |)
+ | WITH (
+ | 'connector' = 'values',
+ | 'readable-metadata' = 'metadata_col:BIGINT, virtual_col:STRING',
+ | 'writable-metadata' = 'metadata_col:BIGINT'
+ |) AS
+ | SELECT
+ | `a`,
+ | `b`
+ | FROM
+ | MyTable
+ |""".stripMargin)
// Same as CTAS
val expected =
