Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[FLINK-36497][table] Remove all deprecated methods in CatalogTable #25930

Closed
wants to merge 2 commits into from
Closed

[FLINK-36497][table] Remove all deprecated methods in CatalogTable #25930

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1a8b33c
[FLINK-36497][table] Remove deprecated method CatalogTable#of
xuyangzhong Jan 8, 2025
07e7449
[FLINK-36497][table] Remove deprecated method CatalogTable#toProperties
xuyangzhong Jan 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions docs/content.zh/docs/dev/table/catalogs.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
Expand Down Expand Up @@ -355,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");
Expand Down
12 changes: 9 additions & 3 deletions docs/content/docs/dev/table/catalogs.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
Expand Down Expand Up @@ -359,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");
Expand Down
7 changes: 6 additions & 1 deletion ...s/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -783,7 +783,12 @@ CatalogBaseTable instantiateCatalogTable(Table hiveTable) {
hiveTable.getViewExpandedText(),
properties);
} else {
return CatalogTable.of(schema, comment, partitionKeys, properties);
return CatalogTable.newBuilder()
.schema(schema)
.comment(comment)
.partitionKeys(partitionKeys)
.options(properties)
.build();
}
}

Expand Down
10 changes: 5 additions & 5 deletions ...ive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserDMLHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,11 +436,11 @@ private ContextResolvedTable createDummyTableForInsertDirectory(
ResolvedSchema resolvedSchema, Map<String, String> props) {
Schema schema = Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();
CatalogTable catalogTable =
CatalogTable.of(
schema,
"a dummy table for the case of insert overwrite directory ",
Collections.emptyList(),
props);
CatalogTable.newBuilder()
.schema(schema)
.comment("a dummy table for the case of insert overwrite directory")
.options(props)
.build();
ResolvedCatalogTable resolvedCatalogTable =
new ResolvedCatalogTable(catalogTable, resolvedSchema);
String currentCatalog = catalogRegistry.getCurrentCatalog();
Expand Down
12 changes: 7 additions & 5 deletions .../java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSemanticAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -662,11 +662,13 @@ private HiveParserASTNode genValuesTempTable(HiveParserASTNode originalFrom, Hiv
ResolvedSchema resolvedSchema = ResolvedSchema.physical(fieldsName, fieldsDataType);
ResolvedCatalogTable tempTable =
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
"values temp table",
new ArrayList<>(),
Collections.emptyMap()),
CatalogTable.newBuilder()
.schema(
Schema.newBuilder()
.fromResolvedSchema(resolvedSchema)
.build())
.comment("values temp table")
.build(),
resolvedSchema);
// remember the data for this table
qb.getValuesTableToData().put(tableName, Tuple2.of(tempTable, valuesData));
Expand Down
48 changes: 31 additions & 17 deletions ...a/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1065,11 +1065,15 @@ private Operation convertCreateTable(HiveParserASTNode ast) throws SemanticExcep
cols, partCols, Collections.emptySet(), null);
ResolvedCatalogTable destTable =
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(schema).build(),
comment,
HiveCatalog.getFieldNames(partCols),
tblProps),
CatalogTable.newBuilder()
.schema(
Schema.newBuilder()
.fromResolvedSchema(schema)
.build())
.comment(comment)
.partitionKeys(HiveCatalog.getFieldNames(partCols))
.options(tblProps)
.build(),
schema);

Tuple4<ObjectIdentifier, QueryOperation, Map<String, String>, Boolean>
Expand Down Expand Up @@ -1192,10 +1196,18 @@ private CreateTableOperation convertCreateTable(
if (uniqueConstraint != null) {
notNullColSet.addAll(uniqueConstraint.getColumns());
}
Schema schema = HiveTableUtil.createSchema(cols, partCols, notNullColSet, uniqueConstraint);
ResolvedSchema schema =
HiveTableUtil.createResolvedSchema(cols, partCols, notNullColSet, uniqueConstraint);
return new CreateTableOperation(
identifier,
CatalogTable.of(schema, comment, HiveCatalog.getFieldNames(partCols), props),
new ResolvedCatalogTable(
CatalogTable.newBuilder()
.schema(Schema.newBuilder().fromResolvedSchema(schema).build())
.comment(comment)
.partitionKeys(HiveCatalog.getFieldNames(partCols))
.options(props)
.build(),
schema),
ifNotExists,
isTemporary);
}
Expand Down Expand Up @@ -1984,11 +1996,12 @@ private Operation convertAlterTableChangeCol(
tableIdentifier,
tableChanges,
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(newSchema).build(),
oldTable.getComment(),
oldTable.getPartitionKeys(),
props),
CatalogTable.newBuilder()
.schema(Schema.newBuilder().fromResolvedSchema(newSchema).build())
.comment(oldTable.getComment())
.partitionKeys(oldTable.getPartitionKeys())
.options(props)
.build(),
newSchema),
false);
}
Expand Down Expand Up @@ -2055,11 +2068,12 @@ private Operation convertAlterTableModifyCols(
return new AlterTableSchemaOperation(
tableIdentifier,
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(newSchema).build(),
oldTable.getComment(),
oldTable.getPartitionKeys(),
props),
CatalogTable.newBuilder()
.schema(Schema.newBuilder().fromResolvedSchema(newSchema).build())
.comment(oldTable.getComment())
.partitionKeys(oldTable.getPartitionKeys())
.options(props)
.build(),
newSchema),
false);
}
Expand Down
20 changes: 9 additions & 11 deletions ...tor-hive/src/test/java/org/apache/flink/connectors/hive/HiveDeserializeExceptionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Properties;

Expand Down Expand Up @@ -74,11 +73,9 @@ public static Object[] parameters() {
new Properties(),
new JobConf(),
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().build(),
null,
new ArrayList<>(),
Collections.emptyMap()),
CatalogTable.newBuilder()
.schema(Schema.newBuilder().build())
.build(),
ResolvedSchema.of()),
HiveShimLoader.getHiveVersion(),
RowType.of(DataTypes.INT().getLogicalType()),
Expand All @@ -87,11 +84,12 @@ public static Object[] parameters() {
ResolvedSchema resolvedSchema = ResolvedSchema.of(Column.physical("i", DataTypes.INT()));
ResolvedCatalogTable resolvedCatalogTable =
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
null,
Collections.emptyList(),
Collections.emptyMap()),
CatalogTable.newBuilder()
.schema(
Schema.newBuilder()
.fromResolvedSchema(resolvedSchema)
.build())
.build(),
resolvedSchema);
HiveSourceBuilder builder =
new HiveSourceBuilder(
Expand Down
26 changes: 15 additions & 11 deletions ...flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveSourceITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,6 @@
import org.junit.BeforeClass;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
Expand Down Expand Up @@ -79,11 +78,13 @@ public void testRegularRead() throws Exception {
hiveCatalog.createTable(
tablePath,
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
null,
new ArrayList<>(),
tableOptions),
CatalogTable.newBuilder()
.schema(
Schema.newBuilder()
.fromResolvedSchema(resolvedSchema)
.build())
.options(tableOptions)
.build(),
resolvedSchema),
false);
HiveTestUtils.createTextTableInserter(
Expand Down Expand Up @@ -126,11 +127,14 @@ public void testRegularRead() throws Exception {
hiveCatalog.createTable(
tablePath,
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
null,
Collections.singletonList("p"),
tableOptions),
CatalogTable.newBuilder()
.schema(
Schema.newBuilder()
.fromResolvedSchema(resolvedSchema)
.build())
.partitionKeys(Collections.singletonList("p"))
.options(tableOptions)
.build(),
resolvedSchema),
false);
HiveTestUtils.createTextTableInserter(
Expand Down
15 changes: 8 additions & 7 deletions ...s/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveSourceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,13 +224,14 @@ private void createTable(ObjectPath tablePath, HiveCatalog hiveCatalog, boolean
hiveCatalog.createTable(
tablePath,
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder()
.fromResolvedSchema(partitionTableRSchema)
.build(),
null,
isPartitioned ? keys : Collections.emptyList(),
tableOptions),
CatalogTable.newBuilder()
.schema(
Schema.newBuilder()
.fromResolvedSchema(partitionTableRSchema)
.build())
.partitionKeys(isPartitioned ? keys : Collections.emptyList())
.options(tableOptions)
.build(),
partitionTableRSchema),
false);

Expand Down
17 changes: 10 additions & 7 deletions ...k-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,6 @@
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
Expand Down Expand Up @@ -88,7 +87,11 @@ void testGenericTable() throws Exception {
Collections.singletonMap(FactoryUtil.CONNECTOR.key(), "COLLECTION");
final CatalogTable table =
new ResolvedCatalogTable(
CatalogTable.of(schema, "csv table", new ArrayList<>(), options),
CatalogTable.newBuilder()
.schema(schema)
.comment("csv table")
.options(options)
.build(),
resolvedSchema);
catalog.createTable(new ObjectPath("mydb", "mytable"), table, true);

Expand Down Expand Up @@ -129,11 +132,11 @@ void testHiveTable() throws Exception {
Collections.singletonMap(FactoryUtil.CONNECTOR.key(), IDENTIFIER);
final CatalogTable table =
new ResolvedCatalogTable(
CatalogTable.of(
Schema.newBuilder().fromResolvedSchema(schema).build(),
"hive table",
new ArrayList<>(),
options),
CatalogTable.newBuilder()
.schema(Schema.newBuilder().fromResolvedSchema(schema).build())
.comment("hive table")
.options(options)
.build(),
schema);
catalog.createTable(new ObjectPath("mydb", "mytable"), table, true);

Expand Down
7 changes: 6 additions & 1 deletion ...or-hive/src/test/java/org/apache/flink/connectors/hive/read/HivePartitionFetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,12 @@ public void testIgnoreNonExistPartition() throws Exception {
List<String> partitionKeys = Collections.singletonList("date");
Map<String, String> options = new HashMap<>();
options.put("connector", "hive");
CatalogTable catalogTable = CatalogTable.of(schema, null, partitionKeys, options);
CatalogTable catalogTable =
CatalogTable.newBuilder()
.schema(schema)
.partitionKeys(partitionKeys)
.options(options)
.build();
ResolvedCatalogTable resolvedCatalogTable =
new ResolvedCatalogTable(
catalogTable, ResolvedSchema.physical(fieldNames, fieldTypes));
Expand Down
21 changes: 11 additions & 10 deletions ...ector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogDataTypeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,16 +205,17 @@ private CatalogTable createCatalogTable(DataType[] types) {
Schema schema = Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();

return new ResolvedCatalogTable(
CatalogTable.of(
schema,
"",
new ArrayList<>(),
new HashMap<String, String>() {
{
put("is_streaming", "false");
put(FactoryUtil.CONNECTOR.key(), IDENTIFIER);
}
}),
CatalogTable.newBuilder()
.schema(schema)
.comment("")
.options(
new HashMap<>() {
{
put("is_streaming", "false");
put(FactoryUtil.CONNECTOR.key(), IDENTIFIER);
}
})
.build(),
resolvedSchema);
}

Expand Down
Loading