You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2025/01/14 04:00:55 UTC
(flink) branch master updated (1f6a6af327d -> 60497e7dfdd)
This is an automated email from the ASF dual-hosted git repository.
ron pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 1f6a6af327d [FLINK-36506][table] Remove all deprecated methods in ColumnStats (#25950)
new 8989f0c1ba3 [FLINK-36497][table] Remove deprecated method CatalogTable#of
new 60497e7dfdd [FLINK-36497][table] Remove deprecated method CatalogTable#toProperties
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content.zh/docs/dev/table/catalogs.md | 12 +-
docs/content/docs/dev/table/catalogs.md | 12 +-
flink-python/pyflink/table/catalog.py | 8 +-
.../table/api/internal/TableEnvironmentImpl.java | 6 +-
.../flink/table/catalog/CatalogTableImpl.java | 119 --------------
.../flink/table/catalog/ConnectorCatalogTable.java | 7 -
.../flink/table/factories/TableFactoryUtil.java | 7 +-
.../table/operations/ddl/CreateTableOperation.java | 8 +-
.../catalog/CatalogBaseTableResolutionTest.java | 7 +-
.../flink/table/catalog/CatalogManagerTest.java | 27 +--
.../flink/table/catalog/CatalogTableImpTest.java | 97 -----------
.../flink/table/catalog/CatalogTestBase.java | 52 +++---
.../resolver/ExpressionResolverTest.java | 14 +-
.../flink/table/operations/QueryOperationTest.java | 16 +-
.../flink/table/catalog/CatalogPropertiesUtil.java | 8 +-
.../apache/flink/table/catalog/CatalogTable.java | 63 +------
.../flink/table/catalog/DefaultCatalogTable.java | 6 -
.../flink/table/catalog/ResolvedCatalogTable.java | 1 -
.../table/legacy/factories/TableSinkFactory.java | 3 +-
.../table/legacy/factories/TableSourceFactory.java | 3 +-
.../flink/table/factories/utils/FactoryMocks.java | 31 ++--
.../table/planner/catalog/CatalogSchemaTable.java | 24 +--
.../table/planner/connectors/DynamicSinkUtils.java | 3 +-
.../operations/SqlCreateTableConverter.java | 4 +-
.../planner/plan/FlinkCalciteCatalogReader.java | 18 +-
.../table/planner/delegation/PlannerBase.scala | 2 +-
.../plan/schema/LegacyCatalogSourceTable.scala | 18 +-
.../planner/catalog/CatalogConstraintTest.java | 12 +-
.../flink/table/planner/catalog/CatalogITCase.java | 4 +-
.../planner/catalog/CatalogStatisticsTest.java | 5 +-
.../planner/catalog/DatabaseCalciteSchemaTest.java | 5 +-
.../planner/catalog/JavaCatalogTableTest.java | 5 -
.../planner/factories/TableFactoryHarness.java | 8 +-
.../planner/factories/TestTimeTravelCatalog.java | 8 +-
.../operations/DeletePushDownUtilsTest.java | 44 ++---
.../operations/SqlDdlToOperationConverterTest.java | 181 ++++++++++-----------
.../operations/SqlDmlToOperationConverterTest.java | 35 ++--
...erializedTableNodeToOperationConverterTest.java | 7 +-
.../SqlNodeToOperationConversionTestBase.java | 2 +-
.../exec/serde/ContextResolvedTableSerdeTest.java | 25 +--
.../exec/serde/DynamicTableSinkSpecSerdeTest.java | 27 ++-
.../serde/DynamicTableSourceSpecSerdeTest.java | 27 ++-
.../serde/TemporalTableSourceSpecSerdeTest.java | 9 +-
.../planner/runtime/batch/sql/RTASITCase.java | 10 +-
.../planner/runtime/stream/sql/RTASITCase.java | 10 +-
.../flink/table/planner/utils/PlannerMocks.java | 12 +-
.../flink/table/api/TableEnvironmentTest.scala | 11 +-
.../planner/plan/metadata/MetadataTestUtil.scala | 44 ++---
.../batch/sql/PartitionableSinkITCase.scala | 29 ++--
.../table/planner/utils/testTableSourceSinks.scala | 18 +-
50 files changed, 430 insertions(+), 684 deletions(-)
delete mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java
delete mode 100644 flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTableImpTest.java
(flink) 02/02: [FLINK-36497][table] Remove deprecated method CatalogTable#toProperties
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 60497e7dfddf27b25b80b913ef259010e97569aa
Author: xuyang <xy...@163.com>
AuthorDate: Thu Jan 9 10:29:49 2025 +0800
[FLINK-36497][table] Remove deprecated method CatalogTable#toProperties
Co-authored-by: shiwei10 <sh...@staff.weibo.com>
---
docs/content.zh/docs/dev/table/catalogs.md | 4 +-
docs/content/docs/dev/table/catalogs.md | 4 +-
flink-python/pyflink/table/catalog.py | 8 +-
.../table/api/internal/TableEnvironmentImpl.java | 6 +-
.../flink/table/catalog/CatalogTableImpl.java | 118 ---------------------
.../flink/table/catalog/ConnectorCatalogTable.java | 7 --
.../flink/table/factories/TableFactoryUtil.java | 7 +-
.../table/operations/ddl/CreateTableOperation.java | 8 +-
.../flink/table/catalog/CatalogTableImpTest.java | 97 -----------------
.../flink/table/catalog/CatalogPropertiesUtil.java | 2 +-
.../apache/flink/table/catalog/CatalogTable.java | 13 ---
.../flink/table/catalog/DefaultCatalogTable.java | 6 --
.../flink/table/catalog/ResolvedCatalogTable.java | 1 -
.../table/legacy/factories/TableSinkFactory.java | 3 +-
.../table/legacy/factories/TableSourceFactory.java | 3 +-
.../table/planner/connectors/DynamicSinkUtils.java | 3 +-
.../operations/SqlCreateTableConverter.java | 4 +-
.../table/planner/delegation/PlannerBase.scala | 2 +-
.../planner/catalog/JavaCatalogTableTest.java | 5 -
.../operations/SqlDdlToOperationConverterTest.java | 8 +-
.../table/planner/utils/testTableSourceSinks.scala | 4 +-
21 files changed, 35 insertions(+), 278 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/catalogs.md b/docs/content.zh/docs/dev/table/catalogs.md
index 0c68ca85cd7..dfbad59df7b 100644
--- a/docs/content.zh/docs/dev/table/catalogs.md
+++ b/docs/content.zh/docs/dev/table/catalogs.md
@@ -361,13 +361,13 @@ catalog.list_databases()
{{< tab "Java/Scala" >}}
```java
// create table
-catalog.createTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
+catalog.createTable(new ObjectPath("mydb", "mytable"), CatalogTable.newBuilder()...build(), false);
// drop table
catalog.dropTable(new ObjectPath("mydb", "mytable"), false);
// alter table
-catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
+catalog.alterTable(new ObjectPath("mydb", "mytable"), CatalogTable.newBuilder()...build(), false);
// rename table
catalog.renameTable(new ObjectPath("mydb", "mytable"), "my_new_table");
diff --git a/docs/content/docs/dev/table/catalogs.md b/docs/content/docs/dev/table/catalogs.md
index 98e324b34f5..851d4f2884a 100644
--- a/docs/content/docs/dev/table/catalogs.md
+++ b/docs/content/docs/dev/table/catalogs.md
@@ -365,13 +365,13 @@ catalog.list_databases()
{{< tab "Java/Scala" >}}
```java
// create table
-catalog.createTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
+catalog.createTable(new ObjectPath("mydb", "mytable"), CatalogTable.newBuilder()...build(), false);
// drop table
catalog.dropTable(new ObjectPath("mydb", "mytable"), false);
// alter table
-catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
+catalog.alterTable(new ObjectPath("mydb", "mytable"), CatalogTable.newBuilder()...build(), false);
// rename table
catalog.renameTable(new ObjectPath("mydb", "mytable"), "my_new_table");
diff --git a/flink-python/pyflink/table/catalog.py b/flink-python/pyflink/table/catalog.py
index f6069a60a1b..4046650025b 100644
--- a/flink-python/pyflink/table/catalog.py
+++ b/flink-python/pyflink/table/catalog.py
@@ -830,8 +830,12 @@ class CatalogBaseTable(object):
gateway = get_gateway()
return CatalogBaseTable(
- gateway.jvm.org.apache.flink.table.catalog.CatalogTableImpl(
- schema._j_table_schema, partition_keys, properties, comment))
+ gateway.jvm.org.apache.flink.table.catalog.CatalogTable.newBuilder()
+ .schema(schema._j_table_schema.toSchema())
+ .comment(comment)
+ .partitionKeys(partition_keys)
+ .options(properties)
+ .build())
@staticmethod
def create_view(
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 4b60d880777..3c807a8a777 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -950,8 +950,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
Catalog catalog =
catalogManager.getCatalogOrThrowException(tableIdentifier.getCatalogName());
- ResolvedCatalogTable catalogTable =
- catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+ ResolvedCatalogTable catalogTable = createTableOperation.getCatalogTable();
Optional<DynamicTableSink> stagingDynamicTableSink =
getSupportsStagingDynamicTableSink(createTableOperation, catalog, catalogTable);
if (stagingDynamicTableSink.isPresent()) {
@@ -985,8 +984,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
ObjectIdentifier tableIdentifier = createTableOperation.getTableIdentifier();
Catalog catalog =
catalogManager.getCatalogOrThrowException(tableIdentifier.getCatalogName());
- ResolvedCatalogTable catalogTable =
- catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable());
+ ResolvedCatalogTable catalogTable = createTableOperation.getCatalogTable();
Optional<DynamicTableSink> stagingDynamicTableSink =
getSupportsStagingDynamicTableSink(createTableOperation, catalog, catalogTable);
if (stagingDynamicTableSink.isPresent()) {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java
deleted file mode 100644
index 61c3ef06cab..00000000000
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java
+++ /dev/null
@@ -1,118 +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.catalog;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.descriptors.DescriptorProperties;
-import org.apache.flink.table.legacy.api.TableSchema;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import static org.apache.flink.table.legacy.descriptors.Schema.SCHEMA;
-
-/**
- * A catalog table implementation.
- *
- * @deprecated Use {@link CatalogTable#newBuilder()} or a custom implementation instead. Don't
- * implement against this internal class. It can lead to unintended side effects if code checks
- * against this class instead of the common interface.
- */
-@Deprecated
-@Internal
-public class CatalogTableImpl extends AbstractCatalogTable {
-
- public CatalogTableImpl(
- TableSchema tableSchema, Map<String, String> properties, String comment) {
- this(tableSchema, new ArrayList<>(), properties, comment);
- }
-
- public CatalogTableImpl(
- TableSchema tableSchema,
- List<String> partitionKeys,
- Map<String, String> properties,
- String comment) {
- super(tableSchema, partitionKeys, properties, comment);
- }
-
- @Override
- public CatalogBaseTable copy() {
- return new CatalogTableImpl(
- getSchema().copy(),
- new ArrayList<>(getPartitionKeys()),
- new HashMap<>(getOptions()),
- getComment());
- }
-
- @Override
- public Optional<String> getDescription() {
- return Optional.of(getComment());
- }
-
- @Override
- public Optional<String> getDetailedDescription() {
- return Optional.of("This is a catalog table in an im-memory catalog");
- }
-
- @Override
- public Map<String, String> toProperties() {
- DescriptorProperties descriptor = new DescriptorProperties(false);
-
- descriptor.putTableSchema(SCHEMA, getSchema());
- descriptor.putPartitionKeys(getPartitionKeys());
-
- Map<String, String> properties = new HashMap<>(getOptions());
-
- descriptor.putProperties(properties);
-
- return descriptor.asMap();
- }
-
- @Override
- public CatalogTable copy(Map<String, String> options) {
- return new CatalogTableImpl(getSchema(), getPartitionKeys(), options, getComment());
- }
-
- /** Construct a {@link CatalogTableImpl} from complete properties that contains table schema. */
- public static CatalogTableImpl fromProperties(Map<String, String> properties) {
- DescriptorProperties descriptorProperties = new DescriptorProperties(false);
- descriptorProperties.putProperties(properties);
- TableSchema tableSchema = descriptorProperties.getTableSchema(SCHEMA);
- List<String> partitionKeys = descriptorProperties.getPartitionKeys();
- return new CatalogTableImpl(
- tableSchema,
- partitionKeys,
- removeRedundant(properties, tableSchema, partitionKeys),
- "");
- }
-
- /** Construct catalog table properties from {@link #toProperties()}. */
- public static Map<String, String> removeRedundant(
- Map<String, String> properties, TableSchema schema, List<String> partitionKeys) {
- Map<String, String> ret = new HashMap<>(properties);
- DescriptorProperties descriptorProperties = new DescriptorProperties(false);
- descriptorProperties.putTableSchema(SCHEMA, schema);
- descriptorProperties.putPartitionKeys(partitionKeys);
- descriptorProperties.asMap().keySet().forEach(ret::remove);
- return ret;
- }
-}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
index 07e48035c0a..352a0dd3c95 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
@@ -93,13 +93,6 @@ public class ConnectorCatalogTable<T1, T2> extends AbstractCatalogTable {
return isBatch;
}
- @Override
- public Map<String, String> toProperties() {
- // This effectively makes sure the table cannot be persisted in a catalog.
- throw new UnsupportedOperationException(
- "ConnectorCatalogTable cannot be converted to properties");
- }
-
@Override
public CatalogTable copy(Map<String, String> options) {
throw new UnsupportedOperationException(
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java
index c27adbd12d4..3b16348a5a6 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java
@@ -28,6 +28,7 @@ import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CommonCatalogOptions;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.listener.CatalogModificationListener;
import org.apache.flink.table.catalog.listener.CatalogModificationListenerFactory;
import org.apache.flink.table.descriptors.ConnectorDescriptorValidator;
@@ -52,7 +53,8 @@ public class TableFactoryUtil {
public static <T> TableSource<T> findAndCreateTableSource(TableSourceFactory.Context context) {
try {
return TableFactoryService.find(
- TableSourceFactory.class, context.getTable().toProperties())
+ TableSourceFactory.class,
+ ((ResolvedCatalogTable) context.getTable()).toProperties())
.createTableSource(context);
} catch (Throwable t) {
throw new TableException("findAndCreateTableSource failed.", t);
@@ -81,7 +83,8 @@ public class TableFactoryUtil {
public static <T> TableSink<T> findAndCreateTableSink(TableSinkFactory.Context context) {
try {
return TableFactoryService.find(
- TableSinkFactory.class, context.getTable().toProperties())
+ TableSinkFactory.class,
+ ((ResolvedCatalogTable) context.getTable()).toProperties())
.createTableSink(context);
} catch (Throwable t) {
throw new TableException("findAndCreateTableSink failed.", t);
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java
index 1e0641e711c..2d3f91710e3 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java
@@ -21,8 +21,8 @@ package org.apache.flink.table.operations.ddl;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.internal.TableResultImpl;
import org.apache.flink.table.api.internal.TableResultInternal;
-import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
@@ -34,13 +34,13 @@ import java.util.Map;
@Internal
public class CreateTableOperation implements CreateOperation {
private final ObjectIdentifier tableIdentifier;
- private final CatalogTable catalogTable;
+ private final ResolvedCatalogTable catalogTable;
private final boolean ignoreIfExists;
private final boolean isTemporary;
public CreateTableOperation(
ObjectIdentifier tableIdentifier,
- CatalogTable catalogTable,
+ ResolvedCatalogTable catalogTable,
boolean ignoreIfExists,
boolean isTemporary) {
this.tableIdentifier = tableIdentifier;
@@ -49,7 +49,7 @@ public class CreateTableOperation implements CreateOperation {
this.isTemporary = isTemporary;
}
- public CatalogTable getCatalogTable() {
+ public ResolvedCatalogTable getCatalogTable() {
return catalogTable;
}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTableImpTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTableImpTest.java
deleted file mode 100644
index 1310e5d292e..00000000000
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTableImpTest.java
+++ /dev/null
@@ -1,97 +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.catalog;
-
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.descriptors.DescriptorProperties;
-import org.apache.flink.table.legacy.api.TableSchema;
-import org.apache.flink.table.legacy.descriptors.Schema;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Test for {@link CatalogTableImpl}. */
-class CatalogTableImpTest {
- private static final String TEST = "test";
-
- @Test
- void testToProperties() {
- TableSchema schema = createTableSchema();
- Map<String, String> prop = createProperties();
- CatalogTable table = new CatalogTableImpl(schema, createPartitionKeys(), prop, TEST);
-
- DescriptorProperties descriptorProperties = new DescriptorProperties(false);
- descriptorProperties.putProperties(table.toProperties());
-
- assertThat(descriptorProperties.getTableSchema(Schema.SCHEMA)).isEqualTo(schema);
- }
-
- @Test
- void testFromProperties() {
- TableSchema schema = createTableSchema();
- Map<String, String> prop = createProperties();
- CatalogTable table = new CatalogTableImpl(schema, createPartitionKeys(), prop, TEST);
-
- CatalogTableImpl tableFromProperties =
- CatalogTableImpl.fromProperties(table.toProperties());
-
- assertThat(table.getOptions()).isEqualTo(tableFromProperties.getOptions());
- assertThat(table.getPartitionKeys()).isEqualTo(tableFromProperties.getPartitionKeys());
- assertThat(table.getSchema()).isEqualTo(tableFromProperties.getSchema());
- }
-
- @Test
- void testNullComment() {
- TableSchema schema = createTableSchema();
- Map<String, String> prop = createProperties();
- CatalogTable table = new CatalogTableImpl(schema, createPartitionKeys(), prop, null);
-
- assertThat(table.getComment()).isEmpty();
- assertThat(table.getDescription()).isEqualTo(Optional.of(""));
- }
-
- private static Map<String, String> createProperties() {
- return new HashMap<String, String>() {
- {
- put("k", "v");
- put("K1", "V1"); // for test case-sensitive
- }
- };
- }
-
- private static TableSchema createTableSchema() {
- return TableSchema.builder()
- .field("first", DataTypes.STRING())
- .field("second", DataTypes.INT())
- .field("third", DataTypes.DOUBLE())
- .field("Fourth", DataTypes.BOOLEAN()) // for test case-sensitive
- .build();
- }
-
- private static List<String> createPartitionKeys() {
- return Arrays.asList("second", "third");
- }
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
index fe2807ce8ee..088870a6db1 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
@@ -236,7 +236,7 @@ public final class CatalogPropertiesUtil {
.schema(schema)
.comment(comment)
.partitionKeys(partitionKeys)
- .options(properties)
+ .options(options)
.snapshot(snapshot)
.build();
} catch (Exception e) {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
index 11712d03c18..5777a1755be 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
@@ -98,19 +98,6 @@ public interface CatalogTable extends CatalogBaseTable {
*/
CatalogTable copy(Map<String, String> options);
- /**
- * Serializes this instance into a map of string-based properties.
- *
- * <p>Compared to the pure table options in {@link #getOptions()}, the map includes schema,
- * partitioning, and other characteristics in a serialized form.
- *
- * @deprecated Only a {@link ResolvedCatalogTable} is serializable to properties.
- */
- @Deprecated
- default Map<String, String> toProperties() {
- return Collections.emptyMap();
- }
-
/** Return the snapshot specified for the table. Return Optional.empty() if not specified. */
default Optional<Long> getSnapshot() {
return Optional.empty();
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
index d2952c48030..b06e6acee26 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
@@ -127,12 +127,6 @@ public class DefaultCatalogTable implements CatalogTable {
return Optional.empty();
}
- @Override
- public Map<String, String> toProperties() {
- throw new UnsupportedOperationException(
- "Only a resolved catalog table can be serialized into a map of string properties.");
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java
index e52f6aa5a13..7f6aba71209 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java
@@ -71,7 +71,6 @@ public final class ResolvedCatalogTable
* symmetric. The framework will resolve functions and perform other validation tasks. A catalog
* implementation must not deal with this during a read operation.
*/
- @Override
public Map<String, String> toProperties() {
return CatalogPropertiesUtil.serializeCatalogTable(this);
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSinkFactory.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSinkFactory.java
index e3294a551d5..9a22bd50e25 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSinkFactory.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSinkFactory.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.legacy.sinks.TableSink;
@@ -65,7 +66,7 @@ public interface TableSinkFactory<T> extends TableFactory {
*/
@Deprecated
default TableSink<T> createTableSink(ObjectPath tablePath, CatalogTable table) {
- return createTableSink(table.toProperties());
+ return createTableSink(((ResolvedCatalogTable) table).toProperties());
}
/**
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSourceFactory.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSourceFactory.java
index 73073f17258..b85fb6eddd5 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSourceFactory.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/legacy/factories/TableSourceFactory.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.legacy.sources.TableSource;
@@ -65,7 +66,7 @@ public interface TableSourceFactory<T> extends TableFactory {
*/
@Deprecated
default TableSource<T> createTableSource(ObjectPath tablePath, CatalogTable table) {
- return createTableSource(table.toProperties());
+ return createTableSource(((ResolvedCatalogTable) table).toProperties());
}
/**
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 285ec39bc52..5af511c83f3 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
@@ -224,8 +224,7 @@ public final class DynamicSinkUtils {
Map<String, String> staticPartitions,
boolean isOverwrite,
DynamicTableSink sink) {
- final ResolvedCatalogTable catalogTable =
- (ResolvedCatalogTable) createTableOperation.getCatalogTable();
+ final ResolvedCatalogTable catalogTable = createTableOperation.getCatalogTable();
final ObjectIdentifier identifier = createTableOperation.getTableIdentifier();
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlCreateTableConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlCreateTableConverter.java
index b45e93f7b85..56129869fc8 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlCreateTableConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlCreateTableConverter.java
@@ -84,7 +84,7 @@ class SqlCreateTableConverter {
/** Convert the {@link SqlCreateTable} node. */
Operation convertCreateTable(SqlCreateTable sqlCreateTable) {
- CatalogTable catalogTable = createCatalogTable(sqlCreateTable);
+ ResolvedCatalogTable catalogTable = createCatalogTable(sqlCreateTable);
UnresolvedIdentifier unresolvedIdentifier =
UnresolvedIdentifier.of(sqlCreateTable.fullTableName());
@@ -188,7 +188,7 @@ class SqlCreateTableConverter {
return catalogManager.resolveCatalogTable(catalogTable);
}
- private CatalogTable createCatalogTable(SqlCreateTable sqlCreateTable) {
+ private ResolvedCatalogTable createCatalogTable(SqlCreateTable sqlCreateTable) {
final Schema sourceTableSchema;
final Optional<TableDistribution> sourceTableDistribution;
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 2254776c9a1..19f50472cb6 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
@@ -528,7 +528,7 @@ abstract class PlannerBase(
staticPartitions: JMap[String, String],
isOverwrite: Boolean): RelNode = {
val input = createRelBuilder.queryOperation(queryOperation).build()
- val table = catalogManager.resolveCatalogTable(createTableOperation.getCatalogTable)
+ val table = createTableOperation.getCatalogTable
val identifier = createTableOperation.getTableIdentifier
val catalog = toScala(catalogManager.getCatalog(identifier.getCatalogName))
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java
index ea36fe66fea..d4c8d1f1aa6 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.java
@@ -285,11 +285,6 @@ class JavaCatalogTableTest extends TableTestBase {
return this;
}
- @Override
- public Map<String, String> toProperties() {
- return Collections.emptyMap();
- }
-
@Override
public Map<String, String> getOptions() {
Map<String, String> map = new HashMap<>();
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
index 8a359d60aed..16c10a4b229 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
@@ -263,7 +263,7 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
Operation operation = parse(sql, planner, parser);
assertThat(operation).isInstanceOf(CreateTableOperation.class);
CreateTableOperation op = (CreateTableOperation) operation;
- CatalogTable catalogTable = op.getCatalogTable();
+ ResolvedCatalogTable catalogTable = op.getCatalogTable();
assertThat(catalogTable.getPartitionKeys()).hasSameElementsAs(Arrays.asList("a", "d"));
assertThat(catalogTable.getSchema().getFieldNames())
.isEqualTo(new String[] {"a", "b", "c", "d"});
@@ -275,9 +275,7 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
DataTypes.INT(),
DataTypes.VARCHAR(Integer.MAX_VALUE)
});
- assertThat(catalogTable).isInstanceOf(ResolvedCatalogTable.class);
- ResolvedCatalogTable resolvedCatalogTable = (ResolvedCatalogTable) catalogTable;
- resolvedCatalogTable
+ catalogTable
.getResolvedSchema()
.getColumn(0)
.ifPresent(
@@ -418,7 +416,7 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
SqlNodeToOperationConversion.convert(planner, catalogManager, node).get();
assertThat(operation).isInstanceOf(CreateTableOperation.class);
CreateTableOperation op = (CreateTableOperation) operation;
- CatalogTable catalogTable = op.getCatalogTable();
+ ResolvedCatalogTable catalogTable = op.getCatalogTable();
Map<String, String> properties = catalogTable.toProperties();
Map<String, String> expected = new HashMap<>();
expected.put("schema.0.name", "a");
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala
index 5bcbf0b9a54..48f61d221a7 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala
@@ -462,11 +462,11 @@ class TestOptionsTableFactory extends TableSourceFactory[Row] with TableSinkFact
}
override def createTableSource(context: TableSourceFactory.Context): TableSource[Row] = {
- createPropertiesSource(context.getTable.toProperties)
+ createPropertiesSource(context.getTable.asInstanceOf[ResolvedCatalogTable].toProperties)
}
override def createTableSink(context: TableSinkFactory.Context): TableSink[Row] = {
- createPropertiesSink(context.getTable.toProperties)
+ createPropertiesSink(context.getTable.asInstanceOf[ResolvedCatalogTable].toProperties)
}
}
(flink) 01/02: [FLINK-36497][table] Remove deprecated method CatalogTable#of
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8989f0c1ba3f7870f79f4a0b5df86d1b4704c47a
Author: xuyang <xy...@163.com>
AuthorDate: Wed Jan 8 20:44:21 2025 +0800
[FLINK-36497][table] Remove deprecated method CatalogTable#of
Co-authored-by: shiwei10 <sh...@staff.weibo.com>
---
docs/content.zh/docs/dev/table/catalogs.md | 8 +-
docs/content/docs/dev/table/catalogs.md | 8 +-
.../flink/table/catalog/CatalogTableImpl.java | 7 +-
.../catalog/CatalogBaseTableResolutionTest.java | 7 +-
.../flink/table/catalog/CatalogManagerTest.java | 27 +---
.../flink/table/catalog/CatalogTestBase.java | 52 ++++---
.../resolver/ExpressionResolverTest.java | 14 +-
.../flink/table/operations/QueryOperationTest.java | 16 +-
.../flink/table/catalog/CatalogPropertiesUtil.java | 8 +-
.../apache/flink/table/catalog/CatalogTable.java | 50 +-----
.../flink/table/factories/utils/FactoryMocks.java | 31 ++--
.../table/planner/catalog/CatalogSchemaTable.java | 24 +--
.../planner/plan/FlinkCalciteCatalogReader.java | 18 ++-
.../plan/schema/LegacyCatalogSourceTable.scala | 18 ++-
.../planner/catalog/CatalogConstraintTest.java | 12 +-
.../flink/table/planner/catalog/CatalogITCase.java | 4 +-
.../planner/catalog/CatalogStatisticsTest.java | 5 +-
.../planner/catalog/DatabaseCalciteSchemaTest.java | 5 +-
.../planner/factories/TableFactoryHarness.java | 8 +-
.../planner/factories/TestTimeTravelCatalog.java | 8 +-
.../operations/DeletePushDownUtilsTest.java | 44 +++---
.../operations/SqlDdlToOperationConverterTest.java | 173 ++++++++++-----------
.../operations/SqlDmlToOperationConverterTest.java | 35 ++---
...erializedTableNodeToOperationConverterTest.java | 7 +-
.../SqlNodeToOperationConversionTestBase.java | 2 +-
.../exec/serde/ContextResolvedTableSerdeTest.java | 25 +--
.../exec/serde/DynamicTableSinkSpecSerdeTest.java | 27 ++--
.../serde/DynamicTableSourceSpecSerdeTest.java | 27 ++--
.../serde/TemporalTableSourceSpecSerdeTest.java | 9 +-
.../planner/runtime/batch/sql/RTASITCase.java | 10 +-
.../planner/runtime/stream/sql/RTASITCase.java | 10 +-
.../flink/table/planner/utils/PlannerMocks.java | 12 +-
.../flink/table/api/TableEnvironmentTest.scala | 11 +-
.../planner/plan/metadata/MetadataTestUtil.scala | 44 +++---
.../batch/sql/PartitionableSinkITCase.scala | 29 ++--
.../table/planner/utils/testTableSourceSinks.scala | 14 +-
36 files changed, 399 insertions(+), 410 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/catalogs.md b/docs/content.zh/docs/dev/table/catalogs.md
index bcfd41231ce..0c68ca85cd7 100644
--- a/docs/content.zh/docs/dev/table/catalogs.md
+++ b/docs/content.zh/docs/dev/table/catalogs.md
@@ -81,7 +81,13 @@ public class MyCatalogSupportTimeTravel implements Catalog {
Map<String, String> options = buildOptions(timestamp);
// Build CatalogTable
CatalogTable catalogTable =
- CatalogTable.of(schema, "", Collections.emptyList(), options, timestamp);
+ CatalogTable.newBuilder()
+ .schema(schema)
+ .comment("")
+ .partitionKeys(Collections.emptyList())
+ .options(options)
+ .snapshot(timestamp)
+ .build();
return catalogTable;
}
}
diff --git a/docs/content/docs/dev/table/catalogs.md b/docs/content/docs/dev/table/catalogs.md
index 534e4be4eaf..98e324b34f5 100644
--- a/docs/content/docs/dev/table/catalogs.md
+++ b/docs/content/docs/dev/table/catalogs.md
@@ -87,7 +87,13 @@ public class MyCatalogSupportTimeTravel implements Catalog {
Map<String, String> options = buildOptions(timestamp);
// Build CatalogTable
CatalogTable catalogTable =
- CatalogTable.of(schema, "", Collections.emptyList(), options, timestamp);
+ CatalogTable.newBuilder()
+ .schema(schema)
+ .comment("")
+ .partitionKeys(Collections.emptyList())
+ .options(options)
+ .snapshot(timestamp)
+ .build();
return catalogTable;
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java
index 5ef6215c471..61c3ef06cab 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.catalog;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.Schema;
import org.apache.flink.table.descriptors.DescriptorProperties;
import org.apache.flink.table.legacy.api.TableSchema;
@@ -34,9 +33,9 @@ import static org.apache.flink.table.legacy.descriptors.Schema.SCHEMA;
/**
* A catalog table implementation.
*
- * @deprecated Use {@link CatalogTable#of(Schema, String, List, Map)} or a custom implementation
- * instead. Don't implement against this internal class. It can lead to unintended side effects
- * if code checks against this class instead of the common interface.
+ * @deprecated Use {@link CatalogTable#newBuilder()} or a custom implementation instead. Don't
+ * implement against this internal class. It can lead to unintended side effects if code checks
+ * against this class instead of the common interface.
*/
@Deprecated
@Internal
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java
index 9746ea9bafb..b56ba696b2a 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java
@@ -341,7 +341,12 @@ class CatalogBaseTableResolutionTest {
options.put("connector", "custom");
options.put("version", "12");
- return CatalogTable.of(TABLE_SCHEMA, comment, partitionKeys, options);
+ return CatalogTable.newBuilder()
+ .schema(TABLE_SCHEMA)
+ .comment(comment)
+ .partitionKeys(partitionKeys)
+ .options(options)
+ .build();
}
private static Map<String, String> catalogTableAsProperties() {
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
index 4f9614d8ac7..97d7f4a3dd1 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
@@ -146,11 +146,7 @@ class CatalogManagerTest {
// Create a table
catalogManager.createTable(
- CatalogTable.of(
- Schema.newBuilder().build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder().schema(Schema.newBuilder().build()).build(),
ObjectIdentifier.of(
catalogManager.getCurrentCatalog(),
catalogManager.getCurrentDatabase(),
@@ -163,11 +159,7 @@ class CatalogManagerTest {
// Create a temporary table
catalogManager.createTemporaryTable(
- CatalogTable.of(
- Schema.newBuilder().build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder().schema(Schema.newBuilder().build()).build(),
ObjectIdentifier.of(
catalogManager.getCurrentCatalog(),
catalogManager.getCurrentDatabase(),
@@ -180,11 +172,10 @@ class CatalogManagerTest {
// Alter a table
catalogManager.alterTable(
- CatalogTable.of(
- Schema.newBuilder().build(),
- "table1 comment",
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().build())
+ .comment("table1 comment")
+ .build(),
ObjectIdentifier.of(
catalogManager.getCurrentCatalog(),
catalogManager.getCurrentDatabase(),
@@ -612,11 +603,7 @@ class CatalogManagerTest {
new CatalogDatabaseImpl(Collections.emptyMap(), "database for exist_cat"),
false);
catalogManager.createTable(
- CatalogTable.of(
- Schema.newBuilder().build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder().schema(Schema.newBuilder().build()).build(),
ObjectIdentifier.of("exist_cat", "cat_db", "test_table"),
false);
catalogManager.createModel(
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTestBase.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTestBase.java
index 0a0b35e3772..53a74f16bef 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTestBase.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogTestBase.java
@@ -55,11 +55,11 @@ public abstract class CatalogTestBase extends CatalogTest {
public CatalogTable createTable() {
final ResolvedSchema resolvedSchema = createSchema();
final CatalogTable origin =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
- TEST_COMMENT,
- Collections.emptyList(),
- getBatchTableProperties());
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .comment(TEST_COMMENT)
+ .options(getBatchTableProperties())
+ .build();
return new ResolvedCatalogTable(origin, resolvedSchema);
}
@@ -87,11 +87,11 @@ public abstract class CatalogTestBase extends CatalogTest {
public CatalogTable createAnotherTable() {
final ResolvedSchema resolvedSchema = createAnotherSchema();
final CatalogTable origin =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
- TEST_COMMENT,
- Collections.emptyList(),
- getBatchTableProperties());
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .comment(TEST_COMMENT)
+ .options(getBatchTableProperties())
+ .build();
return new ResolvedCatalogTable(origin, resolvedSchema);
}
@@ -99,11 +99,11 @@ public abstract class CatalogTestBase extends CatalogTest {
public CatalogTable createStreamingTable() {
final ResolvedSchema resolvedSchema = createSchema();
final CatalogTable origin =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
- TEST_COMMENT,
- Collections.emptyList(),
- getStreamingTableProperties());
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .comment(TEST_COMMENT)
+ .options(getStreamingTableProperties())
+ .build();
return new ResolvedCatalogTable(origin, resolvedSchema);
}
@@ -111,11 +111,12 @@ public abstract class CatalogTestBase extends CatalogTest {
public CatalogTable createPartitionedTable() {
final ResolvedSchema resolvedSchema = createSchema();
final CatalogTable origin =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
- TEST_COMMENT,
- createPartitionKeys(),
- getBatchTableProperties());
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .comment(TEST_COMMENT)
+ .partitionKeys(createPartitionKeys())
+ .options(getBatchTableProperties())
+ .build();
return new ResolvedCatalogTable(origin, resolvedSchema);
}
@@ -123,11 +124,12 @@ public abstract class CatalogTestBase extends CatalogTest {
public CatalogTable createAnotherPartitionedTable() {
final ResolvedSchema resolvedSchema = createAnotherSchema();
final CatalogTable origin =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
- TEST_COMMENT,
- createPartitionKeys(),
- getBatchTableProperties());
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .comment(TEST_COMMENT)
+ .partitionKeys(createPartitionKeys())
+ .options(getBatchTableProperties())
+ .build();
return new ResolvedCatalogTable(origin, resolvedSchema);
}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/resolver/ExpressionResolverTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/resolver/ExpressionResolverTest.java
index 2dece961d4a..21280805a73 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/resolver/ExpressionResolverTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/resolver/ExpressionResolverTest.java
@@ -397,14 +397,12 @@ class ExpressionResolverTest {
new SourceQueryOperation(
ContextResolvedTable.anonymous(
new ResolvedCatalogTable(
- CatalogTable.of(
- schema
- .toSchema(),
- null,
- Collections
- .emptyList(),
- Collections
- .emptyMap()),
+ CatalogTable
+ .newBuilder()
+ .schema(
+ schema
+ .toSchema())
+ .build(),
ResolvedSchema
.physical(
schema
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java
index 064c354dab7..f0aa2f0744c 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java
@@ -53,11 +53,9 @@ class QueryOperationTest {
ContextResolvedTable.temporary(
ObjectIdentifier.of("cat1", "db1", "tab1"),
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder().build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().build())
+ .build(),
schema))),
schema);
@@ -99,11 +97,9 @@ class QueryOperationTest {
ContextResolvedTable.temporary(
ObjectIdentifier.of("cat1", "db1", "tab1"),
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder().build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().build())
+ .build(),
schema))),
schema);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
index 6c374ab0a9e..fe2807ce8ee 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
@@ -232,7 +232,13 @@ public final class CatalogPropertiesUtil {
final Map<String, String> options = deserializeOptions(properties, schemaKey);
- return CatalogTable.of(schema, comment, partitionKeys, options, snapshot);
+ return CatalogTable.newBuilder()
+ .schema(schema)
+ .comment(comment)
+ .partitionKeys(partitionKeys)
+ .options(properties)
+ .snapshot(snapshot)
+ .build();
} catch (Exception e) {
throw new CatalogException("Error in deserializing catalog table.", e);
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
index f3a80342d39..11712d03c18 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
@@ -37,12 +37,12 @@ import java.util.Optional;
* The framework will resolve instances of this interface to a {@link ResolvedCatalogTable} before
* passing it to a {@link DynamicTableFactory} for creating a connector to an external system.
*
- * <p>A catalog implementer can either use {@link #of(Schema, String, List, Map)} for a basic
- * implementation of this interface or create a custom class that allows passing catalog-specific
- * objects all the way down to the connector creation (if necessary).
+ * <p>A catalog implementer can either use {@link #newBuilder()} for a basic implementation of this
+ * interface or create a custom class that allows passing catalog-specific objects all the way down
+ * to the connector creation (if necessary).
*
- * <p>Note: The default implementation that is available via {@link #of(Schema, String, List, Map)}
- * is always serializable. For example, it can be used for implementing a catalog that uses {@link
+ * <p>Note: The default implementation that is available via {@link #newBuilder()} is always
+ * serializable. For example, it can be used for implementing a catalog that uses {@link
* ResolvedCatalogTable#toProperties()} or for persisting compiled plans. An implementation of this
* interface determines whether a catalog table can be serialized by providing a proper {@link
* #getOptions()} method.
@@ -56,46 +56,6 @@ public interface CatalogTable extends CatalogBaseTable {
return new CatalogTable.Builder();
}
- /**
- * Creates a basic implementation of this interface.
- *
- * <p>The signature is similar to a SQL {@code CREATE TABLE} statement.
- *
- * @param schema unresolved schema
- * @param comment optional comment
- * @param partitionKeys list of partition keys or an empty list if not partitioned
- * @param options options to configure the connector
- * @deprecated Use the builder {@link CatalogTable#newBuilder()} instead.
- */
- @Deprecated
- static CatalogTable of(
- Schema schema,
- @Nullable String comment,
- List<String> partitionKeys,
- Map<String, String> options) {
- return new DefaultCatalogTable(schema, comment, partitionKeys, options);
- }
-
- /**
- * Creates an instance of {@link CatalogTable} with a specific snapshot.
- *
- * @param schema unresolved schema
- * @param comment optional comment
- * @param partitionKeys list of partition keys or an empty list if not partitioned
- * @param options options to configure the connector
- * @param snapshot table snapshot of the table
- * @deprecated Use the builder {@link CatalogTable#newBuilder()} instead.
- */
- @Deprecated
- static CatalogTable of(
- Schema schema,
- @Nullable String comment,
- List<String> partitionKeys,
- Map<String, String> options,
- @Nullable Long snapshot) {
- return new DefaultCatalogTable(schema, comment, partitionKeys, options, snapshot, null);
- }
-
/**
* Creates an instance of {@link CatalogTable} from a map of string properties that were
* previously created with {@link ResolvedCatalogTable#toProperties()}.
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/utils/FactoryMocks.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/utils/FactoryMocks.java
index f5362e9abf6..6e8611d65ae 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/utils/FactoryMocks.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/utils/FactoryMocks.java
@@ -65,11 +65,11 @@ public final class FactoryMocks {
null,
IDENTIFIER,
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(schema).build(),
- "mock source",
- Collections.emptyList(),
- options),
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(schema).build())
+ .comment("mock source")
+ .options(options)
+ .build(),
schema),
Collections.emptyMap(),
readableConfig,
@@ -88,11 +88,12 @@ public final class FactoryMocks {
null,
IDENTIFIER,
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(schema).build(),
- "mock sink",
- partitionKeys,
- options),
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(schema).build())
+ .comment("mock source")
+ .partitionKeys(partitionKeys)
+ .options(options)
+ .build(),
schema),
Collections.emptyMap(),
new Configuration(),
@@ -112,11 +113,11 @@ public final class FactoryMocks {
return new FactoryUtil.DefaultDynamicTableContext(
IDENTIFIER,
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(schema).build(),
- "mock context",
- Collections.emptyList(),
- options),
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(schema).build())
+ .comment("mock context")
+ .options(options)
+ .build(),
schema),
enrichmentOptions,
new Configuration(),
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/CatalogSchemaTable.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/CatalogSchemaTable.java
index fcff1aabe80..a9f9472c5f0 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/CatalogSchemaTable.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/CatalogSchemaTable.java
@@ -173,17 +173,19 @@ public class CatalogSchemaTable extends AbstractTable implements TemporalTable {
new TableSourceFactoryContextImpl(
contextResolvedTable.getIdentifier(),
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder()
- .fromResolvedSchema(
- TableSchemaUtils
- .removeTimeAttributeFromResolvedSchema(
- originTable
- .getResolvedSchema()))
- .build(),
- originTable.getComment(),
- originTable.getPartitionKeys(),
- originTable.getOptions()),
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .fromResolvedSchema(
+ TableSchemaUtils
+ .removeTimeAttributeFromResolvedSchema(
+ originTable
+ .getResolvedSchema()))
+ .build())
+ .comment(originTable.getComment())
+ .partitionKeys(originTable.getPartitionKeys())
+ .options(originTable.getOptions())
+ .build(),
originTable.getResolvedSchema()),
config,
contextResolvedTable.isTemporary());
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/FlinkCalciteCatalogReader.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/FlinkCalciteCatalogReader.java
index b8770d2ed90..88e8b080a1c 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/FlinkCalciteCatalogReader.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/FlinkCalciteCatalogReader.java
@@ -237,14 +237,16 @@ public class FlinkCalciteCatalogReader extends CalciteCatalogReader {
TableFactoryUtil.findAndCreateTableSource(
schemaTable.getContextResolvedTable().getIdentifier(),
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder()
- .fromResolvedSchema(
- resolvedSchemaWithRemovedTimeAttribute)
- .build(),
- originTable.getComment(),
- originTable.getPartitionKeys(),
- originTable.getOptions()),
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .fromResolvedSchema(
+ resolvedSchemaWithRemovedTimeAttribute)
+ .build())
+ .comment(originTable.getComment())
+ .partitionKeys(originTable.getPartitionKeys())
+ .options(originTable.getOptions())
+ .build(),
resolvedSchemaWithRemovedTimeAttribute),
new Configuration(),
schemaTable.isTemporary());
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/LegacyCatalogSourceTable.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/LegacyCatalogSourceTable.scala
index b811f21cd27..7f701069e08 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/LegacyCatalogSourceTable.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/LegacyCatalogSourceTable.scala
@@ -182,14 +182,16 @@ class LegacyCatalogSourceTable[T](
val tableSource = TableFactoryUtil.findAndCreateTableSource(
identifier,
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder
- .fromResolvedSchema(resolvedSchemaWithRemovedTimeAttribute)
- .build(),
- tableToFind.getComment,
- tableToFind.getPartitionKeys,
- tableToFind.getOptions
- ),
+ CatalogTable
+ .newBuilder()
+ .schema(
+ Schema.newBuilder
+ .fromResolvedSchema(resolvedSchemaWithRemovedTimeAttribute)
+ .build())
+ .comment(tableToFind.getComment)
+ .partitionKeys(tableToFind.getPartitionKeys)
+ .options(tableToFind.getOptions)
+ .build(),
resolvedSchemaWithRemovedTimeAttribute),
conf,
schemaTable.isTemporary
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogConstraintTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogConstraintTest.java
index 4d5d96e03b7..ac907a04ba4 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogConstraintTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogConstraintTest.java
@@ -80,7 +80,11 @@ public class CatalogConstraintTest {
catalog.createTable(
new ObjectPath(databaseName, "T1"),
- CatalogTable.of(tableSchema, "", Collections.emptyList(), properties),
+ CatalogTable.newBuilder()
+ .schema(tableSchema)
+ .comment("")
+ .options(properties)
+ .build(),
false);
RelNode t1 = TableTestUtil.toRelNode(tEnv.sqlQuery("select * from T1"));
@@ -104,7 +108,11 @@ public class CatalogConstraintTest {
catalog.createTable(
new ObjectPath(databaseName, "T1"),
- CatalogTable.of(tableSchema, "", Collections.emptyList(), properties),
+ CatalogTable.newBuilder()
+ .schema(tableSchema)
+ .comment("")
+ .options(properties)
+ .build(),
false);
RelNode t1 = TableTestUtil.toRelNode(tEnv.sqlQuery("select * from T1"));
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogITCase.java
index eb190a30bbf..8d0914b6fca 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogITCase.java
@@ -44,7 +44,6 @@ import org.junit.jupiter.api.io.TempDir;
import java.net.URLClassLoader;
import java.nio.file.Path;
-import java.util.ArrayList;
import java.util.HashMap;
import static org.assertj.core.api.Assertions.assertThat;
@@ -157,8 +156,7 @@ class CatalogITCase {
catalogManager.registerCatalog("c1", c1);
final CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder().build(), null, new ArrayList<>(), new HashMap<>());
+ CatalogTable.newBuilder().schema(Schema.newBuilder().build()).build();
c1.createDatabase("d1", new CatalogDatabaseImpl(new HashMap<>(), null), true);
c1.createTable(new ObjectPath("d1", "t1"), catalogTable, true);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogStatisticsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogStatisticsTest.java
index 70be9a52c3b..5f25035802e 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogStatisticsTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/CatalogStatisticsTest.java
@@ -58,7 +58,6 @@ import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -124,11 +123,11 @@ class CatalogStatisticsTest {
final Schema schema = Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();
catalog.createTable(
new ObjectPath(databaseName, "T1"),
- CatalogTable.of(schema, "", Collections.emptyList(), properties),
+ CatalogTable.newBuilder().schema(schema).comment("").options(properties).build(),
false);
catalog.createTable(
new ObjectPath(databaseName, "T2"),
- CatalogTable.of(schema, "", Collections.emptyList(), properties),
+ CatalogTable.newBuilder().schema(schema).comment("").options(properties).build(),
false);
alterTableStatistics(catalog, "T1");
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchemaTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchemaTest.java
index 62b4d9bab7b..63f7f3db0e9 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchemaTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchemaTest.java
@@ -29,9 +29,6 @@ import org.apache.flink.table.utils.CatalogManagerMocks;
import org.apache.calcite.schema.Table;
import org.junit.jupiter.api.Test;
-import java.util.ArrayList;
-import java.util.HashMap;
-
import static org.apache.flink.table.utils.CatalogManagerMocks.DEFAULT_CATALOG;
import static org.apache.flink.table.utils.CatalogManagerMocks.DEFAULT_DATABASE;
import static org.assertj.core.api.Assertions.assertThat;
@@ -86,6 +83,6 @@ class DatabaseCalciteSchemaTest {
.primaryKey("a", "b")
.build();
- return CatalogTable.of(schema, null, new ArrayList<>(), new HashMap<>());
+ return CatalogTable.newBuilder().schema(schema).build();
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
index f6eb65bc312..60e62562a8f 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
@@ -306,8 +306,12 @@ public class TableFactoryHarness {
@Override
public CatalogTable copy(Map<String, String> options) {
final CatalogTable parentTable =
- CatalogTable.of(
- getUnresolvedSchema(), getComment(), getPartitionKeys(), options);
+ CatalogTable.newBuilder()
+ .schema(getUnresolvedSchema())
+ .comment(getComment())
+ .partitionKeys(getPartitionKeys())
+ .options(options)
+ .build();
return new HarnessCatalogTable(parentTable, source, sink);
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java
index 6b08c7a1682..8f826b38c20 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestTimeTravelCatalog.java
@@ -29,7 +29,6 @@ import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -77,7 +76,12 @@ public class TestTimeTravelCatalog extends GenericInMemoryCatalog {
String tableName, Schema schema, Map<String, String> properties, long timestamp)
throws TableAlreadyExistException, DatabaseNotExistException, TableNotExistException {
CatalogTable catalogTable =
- CatalogTable.of(schema, "", Collections.emptyList(), properties, timestamp);
+ CatalogTable.newBuilder()
+ .schema(schema)
+ .comment("")
+ .options(properties)
+ .snapshot(timestamp)
+ .build();
ObjectPath objectPath = new ObjectPath(getDefaultDatabase(), tableName);
if (!timeTravelTables.containsKey(objectPath)) {
timeTravelTables.put(objectPath, new ArrayList<>());
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/DeletePushDownUtilsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/DeletePushDownUtilsTest.java
index 7aca5b69f3b..f9c1fd2ba42 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/DeletePushDownUtilsTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/DeletePushDownUtilsTest.java
@@ -124,16 +124,22 @@ public class DeletePushDownUtilsTest {
@Test
public void testGetResolveFilterExpressions() {
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("f0", DataTypes.INT().notNull())
- .column("f1", DataTypes.STRING().nullable())
- .column("f2", DataTypes.BIGINT().nullable())
- .column("f3", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE().nullable())
- .build(),
- null,
- Collections.emptyList(),
- Map.of("connector", TestSimpleDynamicTableSourceFactory.IDENTIFIER()));
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT().notNull())
+ .column("f1", DataTypes.STRING().nullable())
+ .column("f2", DataTypes.BIGINT().nullable())
+ .column(
+ "f3",
+ DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()
+ .nullable())
+ .build())
+ .options(
+ Map.of(
+ "connector",
+ TestSimpleDynamicTableSourceFactory.IDENTIFIER()))
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "t"), false);
@@ -175,15 +181,15 @@ public class DeletePushDownUtilsTest {
}
private CatalogTable createTestCatalogTable(Map<String, String> options) {
- return CatalogTable.of(
- Schema.newBuilder()
- .column("f0", DataTypes.INT().notNull())
- .column("f1", DataTypes.STRING().nullable())
- .column("f2", DataTypes.BIGINT().nullable())
- .build(),
- null,
- Collections.emptyList(),
- options);
+ return CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT().notNull())
+ .column("f1", DataTypes.STRING().nullable())
+ .column("f2", DataTypes.BIGINT().nullable())
+ .build())
+ .options(options)
+ .build();
}
private LogicalTableModify getTableModifyFromSql(String sql) {
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
index 91e8e1d921c..8a359d60aed 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
@@ -441,14 +441,14 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
Map<String, String> sourceProperties = new HashMap<>();
sourceProperties.put("format.type", "json");
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("f0", DataTypes.INT().notNull())
- .column("f1", DataTypes.TIMESTAMP(3))
- .build(),
- null,
- Collections.emptyList(),
- sourceProperties);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT().notNull())
+ .column("f1", DataTypes.TIMESTAMP(3))
+ .build())
+ .options(sourceProperties)
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "sourceTable"), false);
@@ -489,14 +489,14 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
sourceProperties.put("connector.type", "kafka");
sourceProperties.put("format.type", "json");
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("f0", DataTypes.INT().notNull())
- .column("f1", DataTypes.TIMESTAMP(3))
- .build(),
- null,
- Collections.emptyList(),
- sourceProperties);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT().notNull())
+ .column("f1", DataTypes.TIMESTAMP(3))
+ .build())
+ .options(sourceProperties)
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "sourceTable"), false);
final String sql = "create table mytable like `builtin`.`default`.sourceTable";
@@ -521,16 +521,17 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
Map<String, String> sourceProperties = new HashMap<>();
sourceProperties.put("format.type", "json");
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("f0", DataTypes.INT().notNull())
- .column("f1", DataTypes.TIMESTAMP(3))
- .columnByExpression("f2", "`f0` + 12345")
- .watermark("f1", "`f1` - interval '1' second")
- .build(),
- null,
- Arrays.asList("f0", "f1"),
- sourceProperties);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT().notNull())
+ .column("f1", DataTypes.TIMESTAMP(3))
+ .columnByExpression("f2", "`f0` + 12345")
+ .watermark("f1", "`f1` - interval '1' second")
+ .build())
+ .partitionKeys(Arrays.asList("f0", "f1"))
+ .options(sourceProperties)
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "sourceTable"), false);
@@ -702,11 +703,9 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
@Test
public void testCreateTableLikeInvalidPartition() {
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder().column("f0", DataTypes.INT().notNull()).build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap());
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().column("f0", DataTypes.INT().notNull()).build())
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "sourceTable"), false);
@@ -742,11 +741,9 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
@Test
public void testCreateTableLikeInvalidWatermark() {
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder().column("f0", DataTypes.INT().notNull()).build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap());
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().column("f0", DataTypes.INT().notNull()).build())
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "sourceTable"), false);
@@ -768,17 +765,17 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
@Test
public void testCreateTableLikeNestedWatermark() {
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("f0", DataTypes.INT().notNull())
- .column(
- "f1",
- DataTypes.ROW(
- DataTypes.FIELD("tmstmp", DataTypes.TIMESTAMP(3))))
- .build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap());
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT().notNull())
+ .column(
+ "f1",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "tmstmp", DataTypes.TIMESTAMP(3))))
+ .build())
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "sourceTable"), false);
@@ -1227,18 +1224,19 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
// rename column e test computed column expression is ApiExpression which doesn't implement
// the equals method
CatalogTable catalogTable2 =
- CatalogTable.of(
- Schema.newBuilder()
- .column("a", DataTypes.STRING().notNull())
- .column("b", DataTypes.INT().notNull())
- .column("e", DataTypes.STRING())
- .columnByExpression("j", $("e").upperCase())
- .columnByExpression("g", "TO_TIMESTAMP(e)")
- .primaryKey("a", "b")
- .build(),
- "tb2",
- Collections.singletonList("a"),
- Collections.emptyMap());
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("a", DataTypes.STRING().notNull())
+ .column("b", DataTypes.INT().notNull())
+ .column("e", DataTypes.STRING())
+ .columnByExpression("j", $("e").upperCase())
+ .columnByExpression("g", "TO_TIMESTAMP(e)")
+ .primaryKey("a", "b")
+ .build())
+ .comment("tb2")
+ .partitionKeys(Collections.singletonList("a"))
+ .build();
catalogManager
.getCatalog("cat1")
.get()
@@ -2302,18 +2300,19 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
prop.put("connector", "values");
prop.put("bounded", "true");
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("id", DataTypes.INT().notNull())
- .column("measurement", DataTypes.BIGINT().notNull())
- .column(
- "ts",
- DataTypes.ROW(
- DataTypes.FIELD("tmstmp", DataTypes.TIMESTAMP(3))))
- .build(),
- null,
- Collections.emptyList(),
- prop);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT().notNull())
+ .column("measurement", DataTypes.BIGINT().notNull())
+ .column(
+ "ts",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "tmstmp", DataTypes.TIMESTAMP(3))))
+ .build())
+ .options(prop)
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "events"), false);
@@ -2344,14 +2343,14 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
prop.put("connector", "values");
prop.put("bounded", "true");
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("f0", DataTypes.INT())
- .column("f1", DataTypes.VARCHAR(20))
- .build(),
- null,
- Collections.emptyList(),
- prop);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.VARCHAR(20))
+ .build())
+ .options(prop)
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "sourceA"), false);
@@ -2418,14 +2417,14 @@ public class SqlDdlToOperationConverterTest extends SqlNodeToOperationConversion
prop.put("connector", "values");
prop.put("bounded", "true");
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("id", DataTypes.BIGINT().notNull())
- .column("uid", DataTypes.BIGINT().notNull())
- .build(),
- null,
- Collections.emptyList(),
- prop);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT().notNull())
+ .column("uid", DataTypes.BIGINT().notNull())
+ .build())
+ .options(prop)
+ .build();
catalogManager.createTable(
catalogTable, ObjectIdentifier.of("builtin", "default", "id_table"), false);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
index 23b4c2cb98a..057e3127ac1 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDmlToOperationConverterTest.java
@@ -42,7 +42,6 @@ import org.apache.flink.table.planner.parse.CalciteParser;
import org.apache.calcite.sql.SqlNode;
import org.junit.jupiter.api.Test;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -251,14 +250,14 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
Map<String, String> options = new HashMap<>();
options.put("connector", TestUpdateDeleteTableFactory.IDENTIFIER);
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("a", DataTypes.INT().notNull())
- .column("c", DataTypes.STRING().notNull())
- .build(),
- null,
- Collections.emptyList(),
- options);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("a", DataTypes.INT().notNull())
+ .column("c", DataTypes.STRING().notNull())
+ .build())
+ .options(options)
+ .build();
ObjectIdentifier tableIdentifier = ObjectIdentifier.of("builtin", "default", "test_delete");
catalogManager.createTable(catalogTable, tableIdentifier, false);
@@ -285,15 +284,15 @@ public class SqlDmlToOperationConverterTest extends SqlNodeToOperationConversion
Map<String, String> options = new HashMap<>();
options.put("connector", TestUpdateDeleteTableFactory.IDENTIFIER);
CatalogTable catalogTable =
- CatalogTable.of(
- Schema.newBuilder()
- .column("a", DataTypes.INT().notNull())
- .column("b", DataTypes.BIGINT().nullable())
- .column("c", DataTypes.STRING().notNull())
- .build(),
- null,
- Collections.emptyList(),
- options);
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .column("a", DataTypes.INT().notNull())
+ .column("b", DataTypes.BIGINT().nullable())
+ .column("c", DataTypes.STRING().notNull())
+ .build())
+ .options(options)
+ .build();
ObjectIdentifier tableIdentifier = ObjectIdentifier.of("builtin", "default", "test_update");
catalogManager.createTable(catalogTable, tableIdentifier, false);
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 63a91f7e533..9026862e52b 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
@@ -78,7 +78,12 @@ public class SqlMaterializedTableNodeToOperationConverterTest
Map<String, String> options = new HashMap<>();
options.put("connector", "COLLECTION");
final CatalogTable catalogTable =
- CatalogTable.of(tableSchema, "", Arrays.asList("b", "c"), options);
+ CatalogTable.newBuilder()
+ .schema(tableSchema)
+ .comment("")
+ .partitionKeys(Arrays.asList("b", "c"))
+ .options(options)
+ .build();
catalog.createTable(path3, catalogTable, true);
// create materialized table
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java
index 2b87c89ebf8..b8d12188213 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversionTestBase.java
@@ -115,7 +115,7 @@ public class SqlNodeToOperationConversionTestBase {
Map<String, String> options = new HashMap<>();
options.put("connector", "COLLECTION");
final CatalogTable catalogTable =
- CatalogTable.of(tableSchema, "", Collections.emptyList(), options);
+ CatalogTable.newBuilder().schema(tableSchema).comment("").options(options).build();
catalog.createTable(path1, catalogTable, true);
catalog.createTable(path2, catalogTable, true);
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/ContextResolvedTableSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/ContextResolvedTableSerdeTest.java
index a8c8ecbacbd..3494331470e 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/ContextResolvedTableSerdeTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/ContextResolvedTableSerdeTest.java
@@ -295,11 +295,10 @@ public class ContextResolvedTableSerdeTest {
objectIdentifier,
CATALOG,
new ResolvedCatalogTable(
- CatalogTable.of(
- CATALOG_TABLE_SCHEMA,
- null,
- Collections.emptyList(),
- PLAN_OPTIONS),
+ CatalogTable.newBuilder()
+ .schema(CATALOG_TABLE_SCHEMA)
+ .options(PLAN_OPTIONS)
+ .build(),
CATALOG_TABLE_RESOLVED_SCHEMA));
final byte[] actualSerialized =
@@ -417,13 +416,15 @@ public class ContextResolvedTableSerdeTest {
PERMANENT_TABLE_IDENTIFIER,
CATALOG,
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder()
- .fromResolvedSchema(resolvedSchema)
- .build(),
- "my comment",
- PARTITION_KEYS,
- PLAN_OPTIONS),
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .fromResolvedSchema(resolvedSchema)
+ .build())
+ .comment("my comment")
+ .partitionKeys(PARTITION_KEYS)
+ .options(PLAN_OPTIONS)
+ .build(),
resolvedSchema));
final byte[] actualSerialized = createObjectWriter(ctx).writeValueAsBytes(spec);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecSerdeTest.java
index dad7e47eefe..2c1eb026cfe 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecSerdeTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecSerdeTest.java
@@ -88,11 +88,10 @@ class DynamicTableSinkSpecSerdeTest {
Collections.emptyList(),
null);
final CatalogTable catalogTable1 =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema1).build(),
- null,
- Collections.emptyList(),
- options1);
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema1).build())
+ .options(options1)
+ .build();
DynamicTableSinkSpec spec1 =
new DynamicTableSinkSpec(
@@ -119,11 +118,10 @@ class DynamicTableSinkSpecSerdeTest {
Collections.emptyList(),
null);
final CatalogTable catalogTable2 =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema2).build(),
- null,
- Collections.emptyList(),
- options2);
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema2).build())
+ .options(options2)
+ .build();
DynamicTableSinkSpec spec2 =
new DynamicTableSinkSpec(
@@ -156,11 +154,10 @@ class DynamicTableSinkSpecSerdeTest {
Collections.emptyList(),
null);
final CatalogTable catalogTable3 =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema3).build(),
- null,
- Collections.emptyList(),
- options3);
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema3).build())
+ .options(options3)
+ .build();
DynamicTableSinkSpec spec3 =
new DynamicTableSinkSpec(
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java
index 4dbfdd013a9..a483a5a310c 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java
@@ -107,11 +107,10 @@ public class DynamicTableSourceSpecSerdeTest {
null);
final CatalogTable catalogTable1 =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema1).build(),
- null,
- Collections.emptyList(),
- options1);
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema1).build())
+ .options(options1)
+ .build();
DynamicTableSourceSpec spec1 =
new DynamicTableSourceSpec(
@@ -145,11 +144,10 @@ public class DynamicTableSourceSpecSerdeTest {
null);
final CatalogTable catalogTable2 =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema2).build(),
- null,
- Collections.emptyList(),
- options2);
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema2).build())
+ .options(options2)
+ .build();
FlinkTypeFactory factory =
new FlinkTypeFactory(
@@ -366,11 +364,10 @@ public class DynamicTableSourceSpecSerdeTest {
null);
return new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
- null,
- Collections.emptyList(),
- options),
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .options(options)
+ .build(),
resolvedSchema);
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/TemporalTableSourceSpecSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/TemporalTableSourceSpecSerdeTest.java
index 7798e6a94e1..19b48db63c6 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/TemporalTableSourceSpecSerdeTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/TemporalTableSourceSpecSerdeTest.java
@@ -79,11 +79,10 @@ public class TemporalTableSourceSpecSerdeTest {
null);
final CatalogTable catalogTable1 =
- CatalogTable.of(
- Schema.newBuilder().fromResolvedSchema(resolvedSchema1).build(),
- null,
- Collections.emptyList(),
- options1);
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema1).build())
+ .options(options1)
+ .build();
ResolvedCatalogTable resolvedCatalogTable =
new ResolvedCatalogTable(catalogTable1, resolvedSchema1);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RTASITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RTASITCase.java
index 5c02656c5c6..f19039847e7 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RTASITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/RTASITCase.java
@@ -32,7 +32,6 @@ import org.apache.flink.table.types.AbstractDataType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -151,11 +150,10 @@ class RTASITCase extends BatchTestBase {
private CatalogTable getExpectCatalogTable(
String[] cols, AbstractDataType<?>[] fieldDataTypes) {
- return CatalogTable.of(
- Schema.newBuilder().fromFields(cols, fieldDataTypes).build(),
- null,
- Collections.emptyList(),
- getDefaultTargetTableOptions());
+ return CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromFields(cols, fieldDataTypes).build())
+ .options(getDefaultTargetTableOptions())
+ .build();
}
private Map<String, String> getDefaultTargetTableOptions() {
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 61e1c0d087b..c1a943d9f74 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
@@ -32,7 +32,6 @@ import org.apache.flink.table.types.AbstractDataType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -261,11 +260,10 @@ class RTASITCase extends StreamingTestBase {
private CatalogTable getExpectCatalogTable(
String[] cols, AbstractDataType<?>[] fieldDataTypes, Map<String, String> tableOptions) {
- return CatalogTable.of(
- Schema.newBuilder().fromFields(cols, fieldDataTypes).build(),
- null,
- Collections.emptyList(),
- tableOptions);
+ return CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().fromFields(cols, fieldDataTypes).build())
+ .options(tableOptions)
+ .build();
}
private Map<String, String> getDefaultTargetTableOptions() {
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/PlannerMocks.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/PlannerMocks.java
index fe017fdc4b8..c1b77b75122 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/PlannerMocks.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/PlannerMocks.java
@@ -135,11 +135,13 @@ public class PlannerMocks {
public PlannerMocks registerTemporaryTable(String tableName, Schema tableSchema) {
final CatalogTable table =
- CatalogTable.of(
- tableSchema,
- null,
- Collections.emptyList(),
- Map.of("connector", TestSimpleDynamicTableSourceFactory.IDENTIFIER()));
+ CatalogTable.newBuilder()
+ .schema(tableSchema)
+ .options(
+ Map.of(
+ "connector",
+ TestSimpleDynamicTableSourceFactory.IDENTIFIER()))
+ .build();
this.getCatalogManager()
.createTemporaryTable(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
index 78e40e2c28e..2700188b532 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
@@ -2945,11 +2945,12 @@ class TableEnvironmentTest {
table: CatalogBaseTable): CatalogBaseTable = {
numTempTable += 1
if (table.isInstanceOf[CatalogTable]) {
- CatalogTable.of(
- table.getUnresolvedSchema,
- tableComment,
- Collections.emptyList(),
- table.getOptions)
+ CatalogTable
+ .newBuilder()
+ .schema(table.getUnresolvedSchema)
+ .comment(tableComment)
+ .options(table.getOptions)
+ .build()
} else {
val view = table.asInstanceOf[CatalogView]
CatalogView.of(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala
index 654e48f86f3..97cbf1f5ab2 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala
@@ -340,21 +340,21 @@ object MetadataTestUtil {
}
private def createTableSourceTable1(): Table = {
- val catalogTable = CatalogTable.of(
- org.apache.flink.table.api.Schema.newBuilder
- .column("a", DataTypes.BIGINT.notNull)
- .column("b", DataTypes.INT.notNull)
- .column("c", DataTypes.VARCHAR(2147483647).notNull)
- .column("d", DataTypes.BIGINT.notNull)
- .primaryKeyNamed("PK_1", "a", "b")
- .build,
- null,
- Collections.emptyList(),
- Map(
+ val catalogTable = CatalogTable
+ .newBuilder()
+ .schema(
+ org.apache.flink.table.api.Schema.newBuilder
+ .column("a", DataTypes.BIGINT.notNull)
+ .column("b", DataTypes.INT.notNull)
+ .column("c", DataTypes.VARCHAR(2147483647).notNull)
+ .column("d", DataTypes.BIGINT.notNull)
+ .primaryKeyNamed("PK_1", "a", "b")
+ .build)
+ .options(Map(
"connector" -> "values",
"bounded" -> "true"
- )
- )
+ ))
+ .build()
val resolvedSchema = new ResolvedSchema(
util.Arrays.asList(
@@ -468,15 +468,15 @@ object MetadataTestUtil {
}
private def getCatalogTable(resolvedSchema: ResolvedSchema) = {
- CatalogTable.of(
- org.apache.flink.table.api.Schema.newBuilder.fromResolvedSchema(resolvedSchema).build,
- null,
- Collections.emptyList(),
- Map(
- "connector" -> "values",
- "bounded" -> "true"
- )
- )
+ CatalogTable
+ .newBuilder()
+ .schema(org.apache.flink.table.api.Schema.newBuilder.fromResolvedSchema(resolvedSchema).build)
+ .options(
+ Map(
+ "connector" -> "values",
+ "bounded" -> "true"
+ ))
+ .build()
}
private def getMetadataTable(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/PartitionableSinkITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/PartitionableSinkITCase.scala
index d2d17c72278..058c8a360b1 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/PartitionableSinkITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/PartitionableSinkITCase.scala
@@ -317,20 +317,23 @@ object PartitionableSinkITCase {
properties.putString("partition-column." + i, part)
}
- val table = CatalogTable.of(
- Schema
+ val table =
+ CatalogTable
.newBuilder()
- .fromResolvedSchema(
- ResolvedSchema.physical(
- Array("a", "b", "c"),
- TypeConversions.fromLegacyInfoToDataType(rowType.getFieldTypes)
- )
- )
- .build(),
- "",
- util.Arrays.asList[String](partitionColumns: _*),
- properties.asMap()
- )
+ .schema(
+ Schema
+ .newBuilder()
+ .fromResolvedSchema(
+ ResolvedSchema.physical(
+ Array("a", "b", "c"),
+ TypeConversions.fromLegacyInfoToDataType(rowType.getFieldTypes)
+ )
+ )
+ .build())
+ .comment("")
+ .partitionKeys(util.Arrays.asList[String](partitionColumns: _*))
+ .options(properties.asMap())
+ .build()
tEnv
.getCatalog(tEnv.getCurrentCatalog)
.get()
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala
index 008d81f1b84..5bcbf0b9a54 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/testTableSourceSinks.scala
@@ -416,12 +416,14 @@ object TestPartitionableSourceFactory {
}
}
- val table = CatalogTable.of(
- tableSchema,
- "",
- util.Arrays.asList[String]("part1", "part2"),
- properties.asMap()
- )
+ val table =
+ CatalogTable
+ .newBuilder()
+ .schema(tableSchema)
+ .comment("")
+ .partitionKeys(util.Arrays.asList[String]("part1", "part2"))
+ .options(properties.asMap())
+ .build()
val catalog = tEnv.getCatalog(tEnv.getCurrentCatalog).get()
val path = new ObjectPath(tEnv.getCurrentDatabase, tableName)
catalog.createTable(path, table, false)