[lake] Tolerate Paimon lake table existent if the schema and properties matches (apache#1847)

Author: LiebingYu

fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java

@@ -75,6 +75,13 @@ default void close() throws Exception {
     @PublicEvolving
     interface Context {
 
+        /**
+         * Whether the current operation is creating a fluss table.
+         *
+         * @return true if the current operation is creating a fluss table
+         */
+        boolean isCreatingFlussTable();
+
         /** Get the fluss principal currently accessing the catalog. */
         FlussPrincipal getFlussPrincipal();
     }

fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingLakeCatalogContext.java

@@ -22,6 +22,11 @@
 /** A testing implementation of {@link LakeCatalog.Context}. */
 public class TestingLakeCatalogContext implements LakeCatalog.Context {
 
+    @Override
+    public boolean isCreatingFlussTable() {
+        return false;
+    }
+
     @Override
     public FlussPrincipal getFlussPrincipal() {
         return null;

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java

@@ -308,13 +308,9 @@ void testCreateAlreadyExistsLakeTable() throws Exception {
         assertThat(catalog.tableExists(lakeTablePath)).isTrue();
         // drop fluss table
         catalog.dropTable(lakeTablePath, false);
-        // create the table again, should throw exception with ignore if exist = false
-        assertThatThrownBy(() -> catalog.createTable(lakeTablePath, table, false))
-                .isInstanceOf(CatalogException.class)
-                .hasMessage(
-                        String.format(
-                                "The table %s already exists in %s catalog, please first drop the table in %s catalog or use a new table name.",
-                                lakeTablePath, "paimon", "paimon"));
+        assertThat(catalog.tableExists(lakeTablePath)).isFalse();
+        // create the table again should be ok, because the existing lake table is matched
+        catalog.createTable(lakeTablePath, table, false);
     }
 
     @Test

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.lake.paimon;
 
 import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.TableAlreadyExistException;
 import org.apache.fluss.exception.TableNotExistException;
@@ -27,19 +28,24 @@
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.utils.IOUtils;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.CatalogFactory;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypes;
 
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 
+import static org.apache.fluss.lake.paimon.utils.PaimonConversions.FLUSS_CONF_PREFIX;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonSchema;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonSchemaChanges;
@@ -83,12 +89,12 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co
         Identifier paimonPath = toPaimon(tablePath);
         Schema paimonSchema = toPaimonSchema(tableDescriptor);
         try {
-            createTable(paimonPath, paimonSchema);
+            createTable(paimonPath, paimonSchema, context.isCreatingFlussTable());
         } catch (Catalog.DatabaseNotExistException e) {
             // create database
             createDatabase(tablePath.getDatabaseName());
             try {
-                createTable(paimonPath, paimonSchema);
+                createTable(paimonPath, paimonSchema, context.isCreatingFlussTable());
             } catch (Catalog.DatabaseNotExistException t) {
                 // shouldn't happen in normal cases
                 throw new RuntimeException(
@@ -114,13 +120,31 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
         }
     }
 
-    private void createTable(Identifier tablePath, Schema schema)
+    private void createTable(Identifier tablePath, Schema schema, boolean isCreatingFlussTable)
             throws Catalog.DatabaseNotExistException {
         try {
             // not ignore if table exists
             paimonCatalog.createTable(tablePath, schema, false);
         } catch (Catalog.TableAlreadyExistException e) {
-            throw new TableAlreadyExistException("Table " + tablePath + " already exists.");
+            try {
+                Table table = paimonCatalog.getTable(tablePath);
+                FileStoreTable fileStoreTable = (FileStoreTable) table;
+                validatePaimonSchemaCapability(
+                        tablePath, fileStoreTable.schema().toSchema(), schema);
+                // if creating a new fluss table, we should ensure the lake table is empty
+                if (isCreatingFlussTable) {
+                    checkTableIsEmpty(tablePath, fileStoreTable);
+                }
+            } catch (Catalog.TableNotExistException tableNotExistException) {
+                // shouldn't happen in normal cases
+                throw new RuntimeException(
+                        String.format(
+                                "Failed to create table %s in Paimon. The table already existed "
+                                        + "during the initial creation attempt, but subsequently "
+                                        + "could not be found when trying to get it. "
+                                        + "Please check whether the Paimon table was manually deleted, and try again.",
+                                tablePath));
+            }
         }
     }
 
@@ -142,6 +166,41 @@ private void alterTable(Identifier tablePath, List<SchemaChange> tableChanges)
         }
     }
 
+    private void validatePaimonSchemaCapability(
+            Identifier tablePath, Schema existingSchema, Schema newSchema) {
+        // Adjust options for comparison
+        Map<String, String> existingOptions = existingSchema.options();
+        Map<String, String> newOptions = newSchema.options();
+        // `path` will be set automatically by Paimon, so we need to remove it in existing options
+        existingOptions.remove(CoreOptions.PATH.key());
+        // when enable datalake with an existing table, `table.datalake.enabled` will be `false`
+        // in existing options, but `true` in new options.
+        String datalakeConfigKey = FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_ENABLED.key();
+        if (Boolean.FALSE.toString().equalsIgnoreCase(existingOptions.get(datalakeConfigKey))) {
+            existingOptions.remove(datalakeConfigKey);
+            newOptions.remove(datalakeConfigKey);
+        }
+
+        if (!existingSchema.equals(newSchema)) {
+            throw new TableAlreadyExistException(
+                    String.format(
+                            "The table %s already exists in Paimon catalog, but the table schema is not compatible. "
+                                    + "Existing schema: %s, new schema: %s. "
+                                    + "Please first drop the table in Paimon catalog or use a new table name.",
+                            tablePath.getEscapedFullName(), existingSchema, newSchema));
+        }
+    }
+
+    private void checkTableIsEmpty(Identifier tablePath, FileStoreTable table) {
+        if (table.latestSnapshot().isPresent()) {
+            throw new TableAlreadyExistException(
+                    String.format(
+                            "The table %s already exists in Paimon catalog, and the table is not empty. "
+                                    + "Please first drop the table in Paimon catalog or use a new table name.",
+                            tablePath.getEscapedFullName()));
+        }
+    }
+
     @Override
     public void close() {
         IOUtils.closeQuietly(paimonCatalog, "paimon catalog");

…aimon/FlussDataTypeToPaimonDataType.java …utils/FlussDataTypeToPaimonDataType.javafluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/FlussDataTypeToPaimonDataType.java renamed to fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/FlussDataTypeToPaimonDataType.java fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/FlussDataTypeToPaimonDataType.java renamed to fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/FlussDataTypeToPaimonDataType.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.lake.paimon;
+package org.apache.fluss.lake.paimon.utils;
 
 import org.apache.fluss.types.ArrayType;
 import org.apache.fluss.types.BigIntType;

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java

@@ -20,7 +20,6 @@
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.exception.InvalidConfigException;
 import org.apache.fluss.exception.InvalidTableException;
-import org.apache.fluss.lake.paimon.FlussDataTypeToPaimonDataType;
 import org.apache.fluss.lake.paimon.source.FlussRowAsPaimonRow;
 import org.apache.fluss.metadata.TableChange;
 import org.apache.fluss.metadata.TableDescriptor;
@@ -57,7 +56,7 @@ public class PaimonConversions {
     private static final String PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY = "partition.legacy-name";
 
     // for fluss config
-    private static final String FLUSS_CONF_PREFIX = "fluss.";
+    public static final String FLUSS_CONF_PREFIX = "fluss.";
     // for paimon config
     private static final String PAIMON_CONF_PREFIX = "paimon.";
 
@@ -67,6 +66,7 @@ public class PaimonConversions {
     static {
         PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.BUCKET.key());
         PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.BUCKET_KEY.key());
+        PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.PATH.key());
         PAIMON_UNSETTABLE_OPTIONS.add(PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY);
     }