JingsongLi commented on code in PR #230:
URL: https://github.com/apache/flink-table-store/pull/230#discussion_r925542877
##########
flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java:
##########
@@ -68,6 +73,20 @@ public String name() {
return name;
}
+ @Override
+ public void createNamespace(String[] namespace, Map<String, String>
metadata)
+ throws NamespaceAlreadyExistsException {
+ Preconditions.checkArgument(
+ isValidateNamespace(namespace),
+ "Namespace %s is not valid",
+ CatalogV2Implicits.NamespaceHelper(namespace).quoted());
Review Comment:
Don't rely on an internal api `CatalogV2Implicits`.
We can just use `Arrays.toString(...)`.
##########
flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java:
##########
@@ -86,11 +105,18 @@ public String[][] listNamespaces(String[] namespace)
throws NoSuchNamespaceExcep
if (!isValidateNamespace(namespace)) {
throw new NoSuchNamespaceException(namespace);
}
- return new String[0][];
+ List<String> databases = catalog.listDatabases();
+ if (databases.contains(namespace[0])) {
+ return new String[][] {namespace};
+ }
+ throw new NoSuchNamespaceException(namespace);
}
@Override
- public Map<String, String> loadNamespaceMetadata(String[] namespace) {
+ public Map<String, String> loadNamespaceMetadata(String[] namespace)
+ throws NoSuchNamespaceException {
+ // should listNamespace first, because
SupportsNamespace#namespaceExists relies on it
+ listNamespaces(namespace);
Review Comment:
Use `Catalog.databaseExists`?
##########
flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java:
##########
@@ -162,6 +208,18 @@ private SchemaChange toSchemaChange(TableChange change) {
}
}
+ private UpdateSchema toUpdateSchema(
+ StructType schema, Transform[] partitions, Map<String, String>
properties) {
+ return new UpdateSchema(
+ (RowType) toFlinkType(schema),
+ Arrays.stream(partitions)
+ .map(partition -> partition.references()[0].describe())
Review Comment:
check partition complex `Transform` is not supported?
##########
flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java:
##########
@@ -162,6 +208,18 @@ private SchemaChange toSchemaChange(TableChange change) {
}
}
+ private UpdateSchema toUpdateSchema(
+ StructType schema, Transform[] partitions, Map<String, String>
properties) {
+ return new UpdateSchema(
+ (RowType) toFlinkType(schema),
+ Arrays.stream(partitions)
+ .map(partition -> partition.references()[0].describe())
+ .collect(Collectors.toList()),
+ Collections.emptyList(),
Review Comment:
I think we should support table with primary key.
##########
flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java:
##########
@@ -86,11 +105,18 @@ public String[][] listNamespaces(String[] namespace)
throws NoSuchNamespaceExcep
if (!isValidateNamespace(namespace)) {
throw new NoSuchNamespaceException(namespace);
}
- return new String[0][];
+ List<String> databases = catalog.listDatabases();
Review Comment:
I think this should not be the semantic of `listNamespaces`?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
