paimon table ddl

apache · Jan 15, 2025 · e0a6ad0 · e0a6ad0
1 parent 0b1ace7
commit e0a6ad0
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 9 deletions.
diff --git a/...flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkEnvIT.java b/...flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkEnvIT.java
@@ -23,7 +23,6 @@
 import com.google.errorprone.annotations.FormatString;
 import java.io.IOException;
 import java.util.Collections;
-import java.util.Map;
 import java.util.function.Consumer;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.EnvironmentSettings;
@@ -159,18 +158,14 @@ protected TableResult sql(@FormatString String sql, Object... args) {
     return tableEnv.executeSql(String.format(sql, args));
   }
 
-  protected Map<String, String> schemaOptions(String schemaName) {
-    return null;
-  }
-
   protected void doWithSchema(
       Catalog catalog, String schemaName, Consumer<Catalog> action, boolean dropSchema) {
     Preconditions.checkNotNull(catalog);
     Preconditions.checkNotNull(schemaName);
     try {
       tableEnv.useCatalog(catalog.name());
       if (!catalog.asSchemas().schemaExists(schemaName)) {
-        catalog.asSchemas().createSchema(schemaName, null, schemaOptions(schemaName));
+        catalog.asSchemas().createSchema(schemaName, null, null);
       }
       tableEnv.useDatabase(schemaName);
       action.accept(catalog);

diff --git a/...t/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java b/...t/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java
@@ -29,6 +29,7 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.DataTypes;
@@ -587,8 +588,27 @@ protected org.apache.gravitino.Catalog currentCatalog() {
     return hiveCatalog;
   }
 
-  @Override
-  protected Map<String, String> schemaOptions(String schemaName) {
-    return ImmutableMap.of("location", warehouse + "/" + schemaName);
+  protected void doWithSchema(
+      org.apache.gravitino.Catalog catalog,
+      String schemaName,
+      Consumer<org.apache.gravitino.Catalog> action,
+      boolean dropSchema) {
+    Preconditions.checkNotNull(catalog);
+    Preconditions.checkNotNull(schemaName);
+    try {
+      tableEnv.useCatalog(catalog.name());
+      if (!catalog.asSchemas().schemaExists(schemaName)) {
+        catalog
+            .asSchemas()
+            .createSchema(
+                schemaName, null, ImmutableMap.of("location", warehouse + "/" + schemaName));
+      }
+      tableEnv.useDatabase(schemaName);
+      action.accept(catalog);
+    } finally {
+      if (dropSchema) {
+        catalog.asSchemas().dropSchema(schemaName, true);
+      }
+    }
   }
 }