Check for the database instead

apache · Jun 15, 2023 · 4269e50 · 4269e50
1 parent 93a0178
commit 4269e50
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 24 deletions.
diff --git a/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -67,7 +67,6 @@
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
-import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
@@ -127,13 +126,13 @@ public FlinkCatalog(
   @Override
   public void open() throws CatalogException {
     // Create the default database if it does not exist.
-    try {
-      createDatabase(getDefaultDatabase(), ImmutableMap.of(), true);
-    } catch (DatabaseAlreadyExistException e) {
-      // Ignore the exception if it's already exist.
-    } catch (ForbiddenException e) {
-      // Ignore if the user doesn't have CREATE DATABASE permissions,
-      // and we assume that the database already exists
+    String defaultDatabase = getDefaultDatabase();
+    if (!databaseExists(defaultDatabase)) {
+      try {
+        createDatabase(getDefaultDatabase(), ImmutableMap.of(), true);
+      } catch (DatabaseAlreadyExistException e) {
+        // Ignore the exception if it's already exist.
+      }
     }
   }
 

diff --git a/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -67,7 +67,6 @@
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
-import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
@@ -127,13 +126,13 @@ public FlinkCatalog(
   @Override
   public void open() throws CatalogException {
     // Create the default database if it does not exist.
-    try {
-      createDatabase(getDefaultDatabase(), ImmutableMap.of(), true);
-    } catch (DatabaseAlreadyExistException e) {
-      // Ignore the exception if it's already exist.
-    } catch (ForbiddenException e) {
-      // Ignore if the user doesn't have CREATE DATABASE permissions,
-      // and we assume that the database already exists
+    String defaultDatabase = getDefaultDatabase();
+    if (!databaseExists(defaultDatabase)) {
+      try {
+        createDatabase(getDefaultDatabase(), ImmutableMap.of(), true);
+      } catch (DatabaseAlreadyExistException e) {
+        // Ignore the exception if it's already exist.
+      }
     }
   }
 

diff --git a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -67,7 +67,6 @@
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
-import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
@@ -127,13 +126,13 @@ public FlinkCatalog(
   @Override
   public void open() throws CatalogException {
     // Create the default database if it does not exist.
-    try {
-      createDatabase(getDefaultDatabase(), ImmutableMap.of(), true);
-    } catch (DatabaseAlreadyExistException e) {
-      // Ignore the exception if it's already exist.
-    } catch (ForbiddenException e) {
-      // Ignore if the user doesn't have CREATE DATABASE permissions,
-      // and we assume that the database already exists
+    String defaultDatabase = getDefaultDatabase();
+    if (!databaseExists(defaultDatabase)) {
+      try {
+        createDatabase(getDefaultDatabase(), ImmutableMap.of(), true);
+      } catch (DatabaseAlreadyExistException e) {
+        // Ignore the exception if it's already exist.
+      }
     }
   }