FANNG1 commented on code in PR #6362:
URL: https://github.com/apache/gravitino/pull/6362#discussion_r1929912246
##########
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -276,8 +282,21 @@ public void createTable(ObjectPath tablePath,
CatalogBaseTable table, boolean ig
propertiesConverter.toGravitinoTableProperties(table.getOptions());
Transform[] partitions =
partitionConverter.toGravitinoPartitions(((CatalogTable)
table).getPartitionKeys());
+
try {
- catalog().asTableCatalog().createTable(identifier, columns, comment,
properties, partitions);
+
+ Index[] indices = getGrivatinoIndeics(resolvedTable);
Review Comment:
use `getGrivatinoIndices`?
##########
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -521,12 +554,30 @@ protected CatalogBaseTable toFlinkTable(Table table) {
.column(column.name(), column.nullable() ? flinkType.nullable() :
flinkType.notNull())
.withComment(column.comment());
}
+ handleFlinkPrimaryKey(table, builder);
Map<String, String> flinkTableProperties =
propertiesConverter.toFlinkTableProperties(table.properties());
List<String> partitionKeys =
partitionConverter.toFlinkPartitionKeys(table.partitioning());
return CatalogTable.of(builder.build(), table.comment(), partitionKeys,
flinkTableProperties);
}
+ private static void handleFlinkPrimaryKey(
Review Comment:
could you use `String getFlinkPrimaryKey(Table table)` to return the primary
field name, builder could use field name to build primary key.
##########
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -521,12 +554,30 @@ protected CatalogBaseTable toFlinkTable(Table table) {
.column(column.name(), column.nullable() ? flinkType.nullable() :
flinkType.notNull())
.withComment(column.comment());
}
+ handleFlinkPrimaryKey(table, builder);
Map<String, String> flinkTableProperties =
propertiesConverter.toFlinkTableProperties(table.properties());
List<String> partitionKeys =
partitionConverter.toFlinkPartitionKeys(table.partitioning());
return CatalogTable.of(builder.build(), table.comment(), partitionKeys,
flinkTableProperties);
}
+ private static void handleFlinkPrimaryKey(
+ Table table, org.apache.flink.table.api.Schema.Builder builder) {
+ List<Index> primaryKeyList =
+ Arrays.stream(table.index())
+ .filter(index -> index.type() == Index.IndexType.PRIMARY_KEY)
+ .collect(Collectors.toList());
+ if (primaryKeyList.isEmpty()) {
+ return;
+ }
+ Preconditions.checkArgument(
Review Comment:
could you add the check for the size of filednames too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@gravitino.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
