You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "snuyanzin (via GitHub)" <gi...@apache.org> on 2025/09/30 09:48:49 UTC
[PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
snuyanzin opened a new pull request, #27067:
URL: https://github.com/apache/flink/pull/27067
## What is the purpose of the change
The PR doesn't introduce any new feature, however it moves/splits classes in order to make `SchemaConverter` and `ReferenceManager` (functionality for comparing schema related changes) reusable for other objects than `TABLE`
Also it splits AlterTableConverters into separate converters and moves there registration to `SqlNodeConverters`
Table converters are moved to a dedicated package
## Verifying this change
This change is already covered by existing tests
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): ( no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: ( no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? ( no)
- If yes, how is the feature documented? (not applicable )
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #27067:
URL: https://github.com/apache/flink/pull/27067
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390631362
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SchemaConverter.java:
##########
@@ -0,0 +1,327 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters;
+
+import org.apache.flink.sql.parser.ddl.SqlDistribution;
+import org.apache.flink.sql.parser.ddl.SqlTableColumn;
+import org.apache.flink.sql.parser.ddl.SqlWatermark;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.ddl.position.SqlTableColumnPosition;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableDistribution;
+import org.apache.flink.table.expressions.SqlCallExpression;
+import org.apache.flink.table.planner.operations.converters.SqlNodeConverter.ConvertContext;
+import org.apache.flink.table.planner.utils.OperationConverterUtils;
+import org.apache.flink.table.types.AbstractDataType;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlDataTypeSpec;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.planner.calcite.FlinkTypeFactory.toLogicalType;
+import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
+
+/** Base class for schema conversion operations. */
+public abstract class SchemaConverter {
Review Comment:
Extracted from `AlterSchemaConverter` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java#L296
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390662139
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlAlterTableDropColumnConverter.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters.table;
+
+import org.apache.flink.sql.parser.ddl.SqlAlterTableDropColumn;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.planner.operations.converters.SchemaReferencesManager;
+
+import org.apache.calcite.sql.SqlIdentifier;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Convert ALTER TABLE DROP (col1 [, col2, ...]) to generate an updated Schema. */
+public class SqlAlterTableDropColumnConverter
+ extends AbstractAlterTableConverter<SqlAlterTableDropColumn> {
+ @Override
+ protected Operation convertToOperation(
+ SqlAlterTableDropColumn dropColumn,
+ ResolvedCatalogTable oldTable,
+ ConvertContext context) {
+ Set<String> columnsToDrop = new HashSet<>();
+ dropColumn
+ .getColumnList()
+ .forEach(
+ identifier -> {
+ String name = getColumnName((SqlIdentifier) identifier);
+ if (!columnsToDrop.add(name)) {
+ throw new ValidationException(
+ String.format(
+ "%sDuplicate column `%s`.", EX_MSG_PREFIX, name));
+ }
+ });
+
+ SchemaReferencesManager referencesManager = SchemaReferencesManager.create(oldTable);
Review Comment:
The logic extracted from `AlterSchemaConverter` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java#L163
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390635813
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SchemaAddConverter.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters;
+
+import org.apache.flink.sql.parser.ddl.position.SqlTableColumnPosition;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableDistribution;
+import org.apache.flink.table.expressions.SqlCallExpression;
+import org.apache.flink.table.planner.operations.converters.SqlNodeConverter.ConvertContext;
+
+import java.util.Collections;
+import java.util.stream.Collectors;
+
+/** Converter for ALTER TABLE ADD ... schema operations. */
+public class SchemaAddConverter extends SchemaConverter {
Review Comment:
Extracted from `AlterSchemaConverter` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java#L559
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390652553
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractAlterTableConverter.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters.table;
+
+import org.apache.flink.sql.parser.ddl.SqlAlterTable;
+import org.apache.flink.sql.parser.ddl.SqlAlterTableSchema;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableDistribution;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.operations.NopOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.ddl.AlterTableChangeOperation;
+import org.apache.flink.table.operations.utils.ValidationUtils;
+import org.apache.flink.table.planner.operations.converters.SqlNodeConverter;
+import org.apache.flink.table.planner.utils.OperationConverterUtils;
+
+import org.apache.calcite.sql.SqlIdentifier;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Abstract class for ALTER TABLE converters. */
+public abstract class AbstractAlterTableConverter<T extends SqlAlterTable>
+ implements SqlNodeConverter<T> {
+ protected static final String EX_MSG_PREFIX = "Failed to execute ALTER TABLE statement.\n";
+
+ protected abstract Operation convertToOperation(
+ T sqlAlterTable, ResolvedCatalogTable oldTable, ConvertContext context);
+
+ @Override
+ public Operation convertSqlNode(T sqlAlterTable, ConvertContext context) {
+ CatalogManager catalogManager = context.getCatalogManager();
+ final ObjectIdentifier tableIdentifier = getIdentifier(sqlAlterTable, context);
+ Optional<ContextResolvedTable> optionalCatalogTable =
+ catalogManager.getTable(tableIdentifier);
+
+ if (optionalCatalogTable.isEmpty() || optionalCatalogTable.get().isTemporary()) {
+ if (sqlAlterTable.ifTableExists()) {
+ return new NopOperation();
+ }
+ throw new ValidationException(
+ String.format(
+ "Table %s doesn't exist or is a temporary table.", tableIdentifier));
+ }
+ ValidationUtils.validateTableKind(
+ optionalCatalogTable.get().getTable(),
+ CatalogBaseTable.TableKind.TABLE,
+ "alter table");
+
+ return convertToOperation(
+ sqlAlterTable, optionalCatalogTable.get().getResolvedTable(), context);
+ }
Review Comment:
Moved from `SqlNodeToOperationConversion` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java#L393-L410
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390638847
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SchemaModifyConverter.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters;
+
+import org.apache.flink.sql.parser.ddl.SqlTableColumn;
+import org.apache.flink.sql.parser.ddl.position.SqlTableColumnPosition;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableDistribution;
+import org.apache.flink.table.planner.operations.converters.SqlNodeConverter.ConvertContext;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/** Converter for ALTER TABLE MODIFY ... schema operations. */
+public class SchemaModifyConverter extends SchemaConverter {
Review Comment:
Extracted from `AlterSchemaConverter` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java#L661
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27067:
URL: https://github.com/apache/flink/pull/27067#issuecomment-3351046069
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "703a2be24ad7f92e9864ed623521b413f485f275",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "703a2be24ad7f92e9864ed623521b413f485f275",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 703a2be24ad7f92e9864ed623521b413f485f275 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390659353
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java:
##########
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.table.planner.operations.converters;
+package org.apache.flink.table.planner.operations.converters.table;
Review Comment:
table converters moved to `table` package since now there are a lot of them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390662139
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlAlterTableDropColumnConverter.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters.table;
+
+import org.apache.flink.sql.parser.ddl.SqlAlterTableDropColumn;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.planner.operations.converters.SchemaReferencesManager;
+
+import org.apache.calcite.sql.SqlIdentifier;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Convert ALTER TABLE DROP (col1 [, col2, ...]) to generate an updated Schema. */
+public class SqlAlterTableDropColumnConverter
+ extends AbstractAlterTableConverter<SqlAlterTableDropColumn> {
+ @Override
+ protected Operation convertToOperation(
+ SqlAlterTableDropColumn dropColumn,
+ ResolvedCatalogTable oldTable,
+ ConvertContext context) {
+ Set<String> columnsToDrop = new HashSet<>();
+ dropColumn
+ .getColumnList()
+ .forEach(
+ identifier -> {
+ String name = getColumnName((SqlIdentifier) identifier);
+ if (!columnsToDrop.add(name)) {
+ throw new ValidationException(
+ String.format(
+ "%sDuplicate column `%s`.", EX_MSG_PREFIX, name));
+ }
+ });
+
+ SchemaReferencesManager referencesManager = SchemaReferencesManager.create(oldTable);
Review Comment:
The logic extracted from `AlterSchemaConverter`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390643323
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SchemaReferencesManager.java:
##########
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters;
+
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableDistribution;
+import org.apache.flink.table.planner.expressions.ColumnReferenceFinder;
+import org.apache.flink.table.utils.EncodingUtils;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/** Helper class to track column references while (DROP | RENAME) COLUMN operations. */
+public class SchemaReferencesManager {
Review Comment:
Extracted from `AlterSchemaConverter` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java#L774
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390652553
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractAlterTableConverter.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters.table;
+
+import org.apache.flink.sql.parser.ddl.SqlAlterTable;
+import org.apache.flink.sql.parser.ddl.SqlAlterTableSchema;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.TableDistribution;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.operations.NopOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.ddl.AlterTableChangeOperation;
+import org.apache.flink.table.operations.utils.ValidationUtils;
+import org.apache.flink.table.planner.operations.converters.SqlNodeConverter;
+import org.apache.flink.table.planner.utils.OperationConverterUtils;
+
+import org.apache.calcite.sql.SqlIdentifier;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Abstract class for ALTER TABLE converters. */
+public abstract class AbstractAlterTableConverter<T extends SqlAlterTable>
+ implements SqlNodeConverter<T> {
+ protected static final String EX_MSG_PREFIX = "Failed to execute ALTER TABLE statement.\n";
+
+ protected abstract Operation convertToOperation(
+ T sqlAlterTable, ResolvedCatalogTable oldTable, ConvertContext context);
+
+ @Override
+ public Operation convertSqlNode(T sqlAlterTable, ConvertContext context) {
+ CatalogManager catalogManager = context.getCatalogManager();
+ final ObjectIdentifier tableIdentifier = getIdentifier(sqlAlterTable, context);
+ Optional<ContextResolvedTable> optionalCatalogTable =
+ catalogManager.getTable(tableIdentifier);
+
+ if (optionalCatalogTable.isEmpty() || optionalCatalogTable.get().isTemporary()) {
+ if (sqlAlterTable.ifTableExists()) {
+ return new NopOperation();
+ }
+ throw new ValidationException(
+ String.format(
+ "Table %s doesn't exist or is a temporary table.", tableIdentifier));
+ }
+ ValidationUtils.validateTableKind(
+ optionalCatalogTable.get().getTable(),
+ CatalogBaseTable.TableKind.TABLE,
+ "alter table");
+
+ return convertToOperation(
+ sqlAlterTable, optionalCatalogTable.get().getResolvedTable(), context);
+ }
Review Comment:
Moved from `SqlNodeToOperationConversion` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java#L393-L410
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390624738
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java:
##########
@@ -1,1056 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.table.planner.operations;
-
-import org.apache.flink.sql.parser.ddl.SqlAlterTable;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableAdd;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableDropColumn;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableDropConstraint;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableDropPrimaryKey;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableDropWatermark;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableModify;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableRenameColumn;
-import org.apache.flink.sql.parser.ddl.SqlAlterTableSchema;
-import org.apache.flink.sql.parser.ddl.SqlDistribution;
-import org.apache.flink.sql.parser.ddl.SqlTableColumn;
-import org.apache.flink.sql.parser.ddl.SqlWatermark;
-import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
-import org.apache.flink.sql.parser.ddl.position.SqlTableColumnPosition;
-import org.apache.flink.table.api.Schema;
-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.CatalogTable;
-import org.apache.flink.table.catalog.Column;
-import org.apache.flink.table.catalog.ResolvedCatalogTable;
-import org.apache.flink.table.catalog.ResolvedSchema;
-import org.apache.flink.table.catalog.SchemaResolver;
-import org.apache.flink.table.catalog.TableChange;
-import org.apache.flink.table.catalog.TableDistribution;
-import org.apache.flink.table.catalog.UniqueConstraint;
-import org.apache.flink.table.catalog.UnresolvedIdentifier;
-import org.apache.flink.table.expressions.SqlCallExpression;
-import org.apache.flink.table.operations.Operation;
-import org.apache.flink.table.operations.ddl.AlterTableChangeOperation;
-import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
-import org.apache.flink.table.planner.expressions.ColumnReferenceFinder;
-import org.apache.flink.table.planner.utils.OperationConverterUtils;
-import org.apache.flink.table.types.AbstractDataType;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.utils.EncodingUtils;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.sql.SqlDataTypeSpec;
-import org.apache.calcite.sql.SqlIdentifier;
-import org.apache.calcite.sql.SqlNode;
-import org.apache.calcite.sql.validate.SqlValidator;
-
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.function.BiConsumer;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import static org.apache.flink.table.planner.calcite.FlinkTypeFactory.toLogicalType;
-import static org.apache.flink.table.planner.utils.OperationConverterUtils.buildModifyColumnChange;
-import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
-
-/**
- * Converter to convert {@link SqlAlterTableSchema} with source table to generate new {@link
- * Schema}.
- */
-public class AlterSchemaConverter {
Review Comment:
splitted into separate converters
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38458][table] Extract SchemaConverter and ReferenceManager to make them reusable [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27067:
URL: https://github.com/apache/flink/pull/27067#discussion_r2390668427
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/SqlAlterTableDropConstraintConverter.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters.table;
+
+import org.apache.flink.sql.parser.ddl.SqlAlterTableDropConstraint;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.table.operations.Operation;
+
+import org.apache.calcite.sql.SqlIdentifier;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Convert ALTER TABLE DROP CONSTRAINT constraint_name to generate an updated {@link Schema}. */
+public class SqlAlterTableDropConstraintConverter
+ extends AbstractAlterTableConverter<SqlAlterTableDropConstraint> {
+ @Override
+ protected Operation convertToOperation(
+ SqlAlterTableDropConstraint dropConstraint,
+ ResolvedCatalogTable oldTable,
+ ConvertContext context) {
+ Optional<UniqueConstraint> pkConstraint = oldTable.getResolvedSchema().getPrimaryKey();
+ if (pkConstraint.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "%sThe base table does not define any primary key.", EX_MSG_PREFIX));
+ }
+ SqlIdentifier constraintIdentifier = dropConstraint.getConstraintName();
Review Comment:
The logic extracted from `AlterSchemaConverter` https://github.com/apache/flink/blob/538c24c8bf490eb8870c4d39c8ac5b204d3613ee/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java#L238
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org