luoyuxia commented on code in PR #1799:
URL: https://github.com/apache/fluss/pull/1799#discussion_r2426196575
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeCatalog.java:
##########
@@ -17,36 +17,94 @@
package org.apache.fluss.flink.lake;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.CatalogException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.metadata.DataLakeFormat;
+
+import org.apache.flink.table.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.flink.FlinkCatalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.FlinkFileIOLoader;
import org.apache.paimon.options.Options;
+import java.lang.reflect.Method;
+import java.util.HashMap;
import java.util.Map;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
+
/** A lake catalog to delegate the operations on lake table. */
public class LakeCatalog {
+ private static final Map<DataLakeFormat, Catalog> LAKE_CATALOG_CACHE = new
HashMap<>();
+
+ private final String catalogName;
+ private final ClassLoader classLoader;
+
+ public LakeCatalog(String catalogName, ClassLoader classLoader) {
+ this.catalogName = catalogName;
+ this.classLoader = classLoader;
+ }
+
+ public Catalog getLakeCatalog(Configuration tableOptions) {
+ DataLakeFormat lakeFormat =
tableOptions.get(ConfigOptions.TABLE_DATALAKE_FORMAT);
+ Map<String, String> catalogProperties =
Review Comment:
we don't need to always extract lake catalog options, if it's alreay in
cache, we can skip
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeCatalog.java:
##########
@@ -17,36 +17,94 @@
package org.apache.fluss.flink.lake;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.CatalogException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.metadata.DataLakeFormat;
+
+import org.apache.flink.table.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.flink.FlinkCatalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.FlinkFileIOLoader;
import org.apache.paimon.options.Options;
+import java.lang.reflect.Method;
+import java.util.HashMap;
import java.util.Map;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
+
/** A lake catalog to delegate the operations on lake table. */
public class LakeCatalog {
+ private static final Map<DataLakeFormat, Catalog> LAKE_CATALOG_CACHE = new
HashMap<>();
+
+ private final String catalogName;
+ private final ClassLoader classLoader;
+
+ public LakeCatalog(String catalogName, ClassLoader classLoader) {
+ this.catalogName = catalogName;
+ this.classLoader = classLoader;
+ }
+
+ public Catalog getLakeCatalog(Configuration tableOptions) {
+ DataLakeFormat lakeFormat =
tableOptions.get(ConfigOptions.TABLE_DATALAKE_FORMAT);
+ Map<String, String> catalogProperties =
+ DataLakeUtils.extractLakeCatalogProperties(tableOptions);
+ if (lakeFormat == PAIMON && !LAKE_CATALOG_CACHE.containsKey(PAIMON)) {
+ LAKE_CATALOG_CACHE.put(
+ PAIMON,
+ PaimonCatalogFactory.create(catalogName,
catalogProperties, classLoader));
+ } else if (lakeFormat == ICEBERG &&
!LAKE_CATALOG_CACHE.containsKey(ICEBERG)) {
+ LAKE_CATALOG_CACHE.put(
+ ICEBERG, IcebergCatalogFactory.create(catalogName,
catalogProperties));
+ } else {
+ throw new UnsupportedOperationException("Unsupported datalake
format: " + lakeFormat);
+ }
+ return LAKE_CATALOG_CACHE.get(lakeFormat);
Review Comment:
this method is expected to be thread safe since different threads will call
`Catalog#getTable`, but current implemention is not
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkCatalogLakeTest.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.paimon.flink;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.flink.catalog.FlinkCatalog;
+import org.apache.fluss.lake.paimon.testutils.FlinkPaimonTieringTestBase;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
+import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test class for {@link FlinkCatalog}. */
+public class FlinkCatalogLakeTest extends FlinkPaimonTieringTestBase {
Review Comment:
Can we remove this test for paimon since paimon already verify it in
`FlinkUnionReadPrimaryKeyTableITCase`. I want to reduce the test.
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.iceberg.flink;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.flink.catalog.FlinkCatalog;
+import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
+import static org.apache.fluss.lake.iceberg.IcebergLakeCatalog.SYSTEM_COLUMNS;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test class for {@link FlinkCatalog}. */
+public class FlinkCatalogLakeTest extends FlinkIcebergTieringTestBase {
+
+ protected static final String DEFAULT_DB = "fluss";
+
+ protected static final String CATALOG_NAME = "test_iceberg_lake";
+
+ FlinkCatalog catalog;
+
+ @BeforeEach
+ public void beforeEach() {
+ super.beforeEach();
+ buildCatalog();
+ }
+
+ @Test
+ // TODO: duplicate code in paimon and iceberg, refactor it after #1709
Review Comment:
add a todo to remove this test in #1803
I'd like to reduce the number of tests.
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.iceberg.flink;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.flink.catalog.FlinkCatalog;
+import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
+import static org.apache.fluss.lake.iceberg.IcebergLakeCatalog.SYSTEM_COLUMNS;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test class for {@link FlinkCatalog}. */
+public class FlinkCatalogLakeTest extends FlinkIcebergTieringTestBase {
+
+ protected static final String DEFAULT_DB = "fluss";
+
+ protected static final String CATALOG_NAME = "test_iceberg_lake";
+
+ FlinkCatalog catalog;
+
+ @BeforeEach
+ public void beforeEach() {
+ super.beforeEach();
+ buildCatalog();
+ }
+
+ @Test
+ // TODO: duplicate code in paimon and iceberg, refactor it after #1709
Review Comment:
But for this pr, we need to keep this test. Please verify it works for flink
1.18, flink 1.19, flink 2.1(may not exist).
You can just change iceberg-flink-${flink.major.version} in pom, and run the
test manually to verify it.
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.iceberg.flink;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.flink.catalog.FlinkCatalog;
+import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
+import static org.apache.fluss.lake.iceberg.IcebergLakeCatalog.SYSTEM_COLUMNS;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test class for {@link FlinkCatalog}. */
+public class FlinkCatalogLakeTest extends FlinkIcebergTieringTestBase {
Review Comment:
nit:
```suggestion
class FlinkCatalogLakeTest extends FlinkIcebergTieringTestBase {
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeCatalog.java:
##########
@@ -17,36 +17,94 @@
package org.apache.fluss.flink.lake;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.CatalogException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.metadata.DataLakeFormat;
+
+import org.apache.flink.table.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.flink.FlinkCatalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.FlinkFileIOLoader;
import org.apache.paimon.options.Options;
+import java.lang.reflect.Method;
+import java.util.HashMap;
import java.util.Map;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
+
/** A lake catalog to delegate the operations on lake table. */
public class LakeCatalog {
+ private static final Map<DataLakeFormat, Catalog> LAKE_CATALOG_CACHE = new
HashMap<>();
+
+ private final String catalogName;
+ private final ClassLoader classLoader;
+
+ public LakeCatalog(String catalogName, ClassLoader classLoader) {
+ this.catalogName = catalogName;
+ this.classLoader = classLoader;
+ }
+
+ public Catalog getLakeCatalog(Configuration tableOptions) {
+ DataLakeFormat lakeFormat =
tableOptions.get(ConfigOptions.TABLE_DATALAKE_FORMAT);
+ Map<String, String> catalogProperties =
+ DataLakeUtils.extractLakeCatalogProperties(tableOptions);
+ if (lakeFormat == PAIMON && !LAKE_CATALOG_CACHE.containsKey(PAIMON)) {
+ LAKE_CATALOG_CACHE.put(
Review Comment:
we still need to keep the comments:
```
// TODO: Currently, a Fluss cluster only supports a single DataLake storage.
However, in the
// future, it may support multiple DataLakes. The following code
assumes that a single
// lakeCatalog is shared across multiple tables, which will no
longer be valid in such
// cases and should be updated accordingly.
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeCatalog.java:
##########
@@ -17,36 +17,94 @@
package org.apache.fluss.flink.lake;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.CatalogException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.metadata.DataLakeFormat;
+
+import org.apache.flink.table.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.flink.FlinkCatalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.FlinkFileIOLoader;
import org.apache.paimon.options.Options;
+import java.lang.reflect.Method;
+import java.util.HashMap;
import java.util.Map;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
+
/** A lake catalog to delegate the operations on lake table. */
public class LakeCatalog {
+ private static final Map<DataLakeFormat, Catalog> LAKE_CATALOG_CACHE = new
HashMap<>();
Review Comment:
But to use `MapUtils#newConcurrentHashMap`
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeCatalog.java:
##########
@@ -17,36 +17,94 @@
package org.apache.fluss.flink.lake;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.CatalogException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.utils.DataLakeUtils;
+import org.apache.fluss.metadata.DataLakeFormat;
+
+import org.apache.flink.table.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.flink.FlinkCatalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.FlinkFileIOLoader;
import org.apache.paimon.options.Options;
+import java.lang.reflect.Method;
+import java.util.HashMap;
import java.util.Map;
+import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
+import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
+
/** A lake catalog to delegate the operations on lake table. */
public class LakeCatalog {
+ private static final Map<DataLakeFormat, Catalog> LAKE_CATALOG_CACHE = new
HashMap<>();
+
+ private final String catalogName;
+ private final ClassLoader classLoader;
+
+ public LakeCatalog(String catalogName, ClassLoader classLoader) {
+ this.catalogName = catalogName;
+ this.classLoader = classLoader;
+ }
+
+ public Catalog getLakeCatalog(Configuration tableOptions) {
+ DataLakeFormat lakeFormat =
tableOptions.get(ConfigOptions.TABLE_DATALAKE_FORMAT);
+ Map<String, String> catalogProperties =
+ DataLakeUtils.extractLakeCatalogProperties(tableOptions);
+ if (lakeFormat == PAIMON && !LAKE_CATALOG_CACHE.containsKey(PAIMON)) {
+ LAKE_CATALOG_CACHE.put(
+ PAIMON,
+ PaimonCatalogFactory.create(catalogName,
catalogProperties, classLoader));
+ } else if (lakeFormat == ICEBERG &&
!LAKE_CATALOG_CACHE.containsKey(ICEBERG)) {
+ LAKE_CATALOG_CACHE.put(
+ ICEBERG, IcebergCatalogFactory.create(catalogName,
catalogProperties));
+ } else {
+ throw new UnsupportedOperationException("Unsupported datalake
format: " + lakeFormat);
+ }
+ return LAKE_CATALOG_CACHE.get(lakeFormat);
+ }
+
+ /**
+ * Factory for creating Paimon Catalog instances.
+ *
+ * <p>Purpose: Encapsulates Paimon-related dependencies (e.g.
FlinkFileIOLoader) to avoid direct
+ * dependency in the main LakeCatalog class.
+ */
+ public static class PaimonCatalogFactory {
+
+ private PaimonCatalogFactory() {}
- // currently, only support paimon
- // todo make it pluggable
- private final FlinkCatalog paimonFlinkCatalog;
-
- public LakeCatalog(
- String catalogName, Map<String, String> catalogProperties,
ClassLoader classLoader) {
- CatalogContext catalogContext =
- CatalogContext.create(
- Options.fromMap(catalogProperties), null, new
FlinkFileIOLoader());
- paimonFlinkCatalog =
- FlinkCatalogFactory.createCatalog(catalogName, catalogContext,
classLoader);
+ public static Catalog create(
+ String catalogName, Map<String, String> properties,
ClassLoader classLoader) {
+ return FlinkCatalogFactory.createCatalog(
+ catalogName,
+ CatalogContext.create(
+ Options.fromMap(properties), null, new
FlinkFileIOLoader()),
+ classLoader);
+ }
}
- public CatalogBaseTable getTable(ObjectPath objectPath)
- throws TableNotExistException, CatalogException {
- return paimonFlinkCatalog.getTable(objectPath);
+ /** Factory use reflection to create Iceberg Catalog instances. */
+ public static class IcebergCatalogFactory {
+
+ private IcebergCatalogFactory() {}
+
+ public static Catalog create(String catalogName, Map<String, String>
properties) {
+ properties.put("catalog-type", properties.get("type"));
Review Comment:
`type` may be null if user use `catalog-impl` to configure iceberg in Fluss
cluster side.
--
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]
