You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2026/02/19 18:38:31 UTC
(flink) branch master updated: [FLINK-39029] move catalog store factory builder to new class (#27531)
This is an automated email from the ASF dual-hosted git repository.
rmetzger pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b1192e8a8e8 [FLINK-39029] move catalog store factory builder to new class (#27531)
b1192e8a8e8 is described below
commit b1192e8a8e8d2756388394acf42d58a38b419cd3
Author: Hao Li <11...@users.noreply.github.com>
AuthorDate: Thu Feb 19 10:38:22 2026 -0800
[FLINK-39029] move catalog store factory builder to new class (#27531)
* [FLINK-39029] move catalog store factory builder to new class
---
.gitignore | 1 +
.../gateway/service/context/SessionContext.java | 5 +-
.../java/internal/StreamTableEnvironmentImpl.java | 17 +--
.../flink/table/api/EnvironmentSettings.java | 5 +-
.../table/api/internal/TableEnvironmentImpl.java | 17 +--
.../flink/table/factories/ApiFactoryUtil.java | 130 ++++++++++++++++++++
.../flink/table/factories/TableFactoryUtil.java | 57 +--------
.../flink/table/factories/ApiFactoryUtilTest.java | 136 +++++++++++++++++++++
.../internal/StreamTableEnvironmentImpl.scala | 23 ++--
9 files changed, 304 insertions(+), 87 deletions(-)
diff --git a/.gitignore b/.gitignore
index 7f204c296f3..0ed273dea3d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,6 +7,7 @@ scalastyle-output.xml
!.idea/vcs.xml
!.idea/icon.png
.vscode
+.claude
.metals
.bloop
.cursor
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
index 12bfc2d8e5a..b5463d44267 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
@@ -31,6 +31,7 @@ import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.CatalogStoreHolder;
import org.apache.flink.table.catalog.FunctionCatalog;
import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.factories.ApiFactoryUtil;
import org.apache.flink.table.factories.CatalogStoreFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.TableFactoryUtil;
@@ -390,9 +391,9 @@ public class SessionContext {
SessionEnvironment environment) {
CatalogStoreFactory catalogStoreFactory =
- TableFactoryUtil.findAndCreateCatalogStoreFactory(configuration, userClassLoader);
+ ApiFactoryUtil.findAndCreateCatalogStoreFactory(configuration, userClassLoader);
CatalogStoreFactory.Context catalogStoreFactoryContext =
- TableFactoryUtil.buildCatalogStoreFactoryContext(configuration, userClassLoader);
+ ApiFactoryUtil.buildCatalogStoreFactoryContext(configuration, userClassLoader);
catalogStoreFactory.open(catalogStoreFactoryContext);
CatalogStoreHolder catalogStore =
CatalogStoreHolder.newBuilder()
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
index 2bba1d76d69..810546ed6a6 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
@@ -46,6 +46,7 @@ import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.delegation.Executor;
import org.apache.flink.table.delegation.Planner;
import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.ApiFactoryUtil;
import org.apache.flink.table.factories.CatalogStoreFactory;
import org.apache.flink.table.factories.PlannerFactoryUtil;
import org.apache.flink.table.factories.TableFactoryUtil;
@@ -112,17 +113,11 @@ public final class StreamTableEnvironmentImpl extends AbstractStreamTableEnviron
new ResourceManager(settings.getConfiguration(), userClassLoader);
final ModuleManager moduleManager = new ModuleManager();
- final CatalogStoreFactory catalogStoreFactory =
- TableFactoryUtil.findAndCreateCatalogStoreFactory(
- settings.getConfiguration(), userClassLoader);
- final CatalogStoreFactory.Context catalogStoreFactoryContext =
- TableFactoryUtil.buildCatalogStoreFactoryContext(
- settings.getConfiguration(), userClassLoader);
- catalogStoreFactory.open(catalogStoreFactoryContext);
- final CatalogStore catalogStore =
- settings.getCatalogStore() != null
- ? settings.getCatalogStore()
- : catalogStoreFactory.createCatalogStore();
+ final ApiFactoryUtil.CatalogStoreResult catalogStoreResult =
+ ApiFactoryUtil.getOrCreateCatalogStore(
+ settings.getCatalogStore(), settings.getConfiguration(), userClassLoader);
+ final CatalogStore catalogStore = catalogStoreResult.getCatalogStore();
+ final CatalogStoreFactory catalogStoreFactory = catalogStoreResult.getCatalogStoreFactory();
final CatalogManager catalogManager =
CatalogManager.newBuilder()
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
index 67fcfcec463..e08704a9b99 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
@@ -144,9 +144,8 @@ public class EnvironmentSettings {
}
@Internal
- @Nullable
- public CatalogStore getCatalogStore() {
- return catalogStore;
+ public Optional<CatalogStore> getCatalogStore() {
+ return Optional.ofNullable(catalogStore);
}
@Internal
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 6a40fa9c238..388b4f969db 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -83,6 +83,7 @@ import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.ModelReferenceExpression;
import org.apache.flink.table.expressions.TableReferenceExpression;
import org.apache.flink.table.expressions.utils.ApiExpressionDefaultVisitor;
+import org.apache.flink.table.factories.ApiFactoryUtil;
import org.apache.flink.table.factories.CatalogStoreFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.PlannerFactoryUtil;
@@ -251,17 +252,11 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
userClassLoader, ExecutorFactory.class, ExecutorFactory.DEFAULT_IDENTIFIER);
final Executor executor = executorFactory.create(settings.getConfiguration());
- final CatalogStoreFactory catalogStoreFactory =
- TableFactoryUtil.findAndCreateCatalogStoreFactory(
- settings.getConfiguration(), userClassLoader);
- final CatalogStoreFactory.Context context =
- TableFactoryUtil.buildCatalogStoreFactoryContext(
- settings.getConfiguration(), userClassLoader);
- catalogStoreFactory.open(context);
- final CatalogStore catalogStore =
- settings.getCatalogStore() != null
- ? settings.getCatalogStore()
- : catalogStoreFactory.createCatalogStore();
+ final ApiFactoryUtil.CatalogStoreResult catalogStoreResult =
+ ApiFactoryUtil.getOrCreateCatalogStore(
+ settings.getCatalogStore(), settings.getConfiguration(), userClassLoader);
+ final CatalogStore catalogStore = catalogStoreResult.getCatalogStore();
+ final CatalogStoreFactory catalogStoreFactory = catalogStoreResult.getCatalogStoreFactory();
// use configuration to init table config
final TableConfig tableConfig = TableConfig.getDefault();
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/ApiFactoryUtil.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/ApiFactoryUtil.java
new file mode 100644
index 00000000000..cb3b9189d46
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/ApiFactoryUtil.java
@@ -0,0 +1,130 @@
+/*
+ * 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.flink.table.factories;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DelegatingConfiguration;
+import org.apache.flink.table.catalog.CatalogStore;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.Optional;
+
+/** Utility for dealing with catalog store factories. */
+@Internal
+public class ApiFactoryUtil {
+
+ /** Result holder for catalog store and factory. */
+ @Internal
+ public static class CatalogStoreResult {
+ private final CatalogStore catalogStore;
+ @Nullable private final CatalogStoreFactory catalogStoreFactory;
+
+ public CatalogStoreResult(
+ CatalogStore catalogStore, @Nullable CatalogStoreFactory catalogStoreFactory) {
+ this.catalogStore = catalogStore;
+ this.catalogStoreFactory = catalogStoreFactory;
+ }
+
+ public CatalogStore getCatalogStore() {
+ return catalogStore;
+ }
+
+ @Nullable
+ public CatalogStoreFactory getCatalogStoreFactory() {
+ return catalogStoreFactory;
+ }
+ }
+
+ /**
+ * Gets or creates a {@link CatalogStore}. If a catalog store is provided in settings, it will
+ * be used directly. Otherwise, a new catalog store will be created using the factory.
+ *
+ * @param providedCatalogStore the catalog store from settings, if present
+ * @param configuration the configuration
+ * @param classLoader the user classloader
+ * @return a result containing the catalog store and factory (factory is null if store was
+ * provided)
+ */
+ public static CatalogStoreResult getOrCreateCatalogStore(
+ Optional<CatalogStore> providedCatalogStore,
+ Configuration configuration,
+ ClassLoader classLoader) {
+ if (providedCatalogStore.isPresent()) {
+ return new CatalogStoreResult(providedCatalogStore.get(), null);
+ } else {
+ CatalogStoreFactory catalogStoreFactory =
+ findAndCreateCatalogStoreFactory(configuration, classLoader);
+ CatalogStoreFactory.Context catalogStoreFactoryContext =
+ buildCatalogStoreFactoryContext(configuration, classLoader);
+ catalogStoreFactory.open(catalogStoreFactoryContext);
+ CatalogStore catalogStore = catalogStoreFactory.createCatalogStore();
+ return new CatalogStoreResult(catalogStore, catalogStoreFactory);
+ }
+ }
+
+ /**
+ * Finds and creates a {@link CatalogStoreFactory} using the provided {@link Configuration} and
+ * user classloader.
+ *
+ * <p>The configuration format should be as follows:
+ *
+ * <pre>{@code
+ * table.catalog-store.kind: {identifier}
+ * table.catalog-store.{identifier}.{param1}: xxx
+ * table.catalog-store.{identifier}.{param2}: xxx
+ * }</pre>
+ */
+ public static CatalogStoreFactory findAndCreateCatalogStoreFactory(
+ Configuration configuration, ClassLoader classLoader) {
+ String identifier = configuration.get(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND);
+
+ CatalogStoreFactory catalogStoreFactory =
+ FactoryUtil.discoverFactory(classLoader, CatalogStoreFactory.class, identifier);
+
+ return catalogStoreFactory;
+ }
+
+ /**
+ * Build a {@link CatalogStoreFactory.Context} for opening the {@link CatalogStoreFactory}.
+ *
+ * <p>The configuration format should be as follows:
+ *
+ * <pre>{@code
+ * table.catalog-store.kind: {identifier}
+ * table.catalog-store.{identifier}.{param1}: xxx
+ * table.catalog-store.{identifier}.{param2}: xxx
+ * }</pre>
+ */
+ public static CatalogStoreFactory.Context buildCatalogStoreFactoryContext(
+ Configuration configuration, ClassLoader classLoader) {
+ String identifier = configuration.get(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND);
+ String catalogStoreOptionPrefix =
+ CommonCatalogOptions.TABLE_CATALOG_STORE_OPTION_PREFIX + identifier + ".";
+ Map<String, String> options =
+ new DelegatingConfiguration(configuration, catalogStoreOptionPrefix).toMap();
+ CatalogStoreFactory.Context context =
+ new FactoryUtil.DefaultCatalogStoreContext(options, configuration, classLoader);
+
+ return context;
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java
index ad681b54394..09e3951fe40 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java
@@ -19,14 +19,11 @@
package org.apache.flink.table.factories;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.DelegatingConfiguration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogTable;
-import org.apache.flink.table.catalog.CommonCatalogOptions;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.listener.CatalogModificationListener;
@@ -42,10 +39,14 @@ import org.apache.flink.table.legacy.sources.TableSource;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.stream.Collectors;
-/** Utility for dealing with {@link TableFactory} using the {@link TableFactoryService}. */
+/**
+ * Utility for dealing with {@link TableFactory} using the {@link TableFactoryService}.
+ *
+ * @deprecated Use {@link FactoryUtil} instead.
+ */
+@Deprecated
@Internal
public class TableFactoryUtil {
@@ -174,50 +175,4 @@ public class TableFactoryUtil {
}))
.collect(Collectors.toList());
}
-
- /**
- * Finds and creates a {@link CatalogStoreFactory} using the provided {@link Configuration} and
- * user classloader.
- *
- * <p>The configuration format should be as follows:
- *
- * <pre>{@code
- * table.catalog-store.kind: {identifier}
- * table.catalog-store.{identifier}.{param1}: xxx
- * table.catalog-store.{identifier}.{param2}: xxx
- * }</pre>
- */
- public static CatalogStoreFactory findAndCreateCatalogStoreFactory(
- Configuration configuration, ClassLoader classLoader) {
- String identifier = configuration.get(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND);
-
- CatalogStoreFactory catalogStoreFactory =
- FactoryUtil.discoverFactory(classLoader, CatalogStoreFactory.class, identifier);
-
- return catalogStoreFactory;
- }
-
- /**
- * Build a {@link CatalogStoreFactory.Context} for opening the {@link CatalogStoreFactory}.
- *
- * <p>The configuration format should be as follows:
- *
- * <pre>{@code
- * table.catalog-store.kind: {identifier}
- * table.catalog-store.{identifier}.{param1}: xxx
- * table.catalog-store.{identifier}.{param2}: xxx
- * }</pre>
- */
- public static CatalogStoreFactory.Context buildCatalogStoreFactoryContext(
- Configuration configuration, ClassLoader classLoader) {
- String identifier = configuration.get(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND);
- String catalogStoreOptionPrefix =
- CommonCatalogOptions.TABLE_CATALOG_STORE_OPTION_PREFIX + identifier + ".";
- Map<String, String> options =
- new DelegatingConfiguration(configuration, catalogStoreOptionPrefix).toMap();
- CatalogStoreFactory.Context context =
- new FactoryUtil.DefaultCatalogStoreContext(options, configuration, classLoader);
-
- return context;
- }
}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/factories/ApiFactoryUtilTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/factories/ApiFactoryUtilTest.java
new file mode 100644
index 00000000000..c8e1908291b
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/factories/ApiFactoryUtilTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.flink.table.factories;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.apache.flink.table.catalog.FileCatalogStoreFactory;
+import org.apache.flink.table.catalog.GenericInMemoryCatalogStoreFactory;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.File;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ApiFactoryUtil}. */
+class ApiFactoryUtilTest {
+
+ @ParameterizedTest(name = "kind={0}, expectedFactory={1}")
+ @MethodSource("catalogStoreFactoryTestParameters")
+ void testFindAndCreateCatalogStoreFactory(String kind, Class<?> expectedFactoryClass) {
+ Configuration configuration = new Configuration();
+ if (kind != null) {
+ configuration.set(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND, kind);
+ }
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+
+ CatalogStoreFactory factory =
+ ApiFactoryUtil.findAndCreateCatalogStoreFactory(configuration, classLoader);
+
+ assertThat(factory).isInstanceOf(expectedFactoryClass);
+ }
+
+ @Test
+ void testBuildCatalogStoreFactoryContext(@TempDir File tempFolder) {
+ Configuration configuration = new Configuration();
+ configuration.set(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND, "file");
+ configuration.setString("table.catalog-store.file.path", tempFolder.getAbsolutePath());
+ configuration.setString("table.catalog-store.file.option1", "value1");
+ configuration.setString("table.catalog-store.file.option2", "value2");
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+
+ CatalogStoreFactory.Context context =
+ ApiFactoryUtil.buildCatalogStoreFactoryContext(configuration, classLoader);
+
+ assertThat(context).isNotNull();
+ assertThat(context.getOptions())
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(
+ "path", tempFolder.getAbsolutePath(),
+ "option1", "value1",
+ "option2", "value2"));
+ assertThat(context.getConfiguration()).isEqualTo(configuration);
+ assertThat(context.getClassLoader()).isEqualTo(classLoader);
+ }
+
+ @Test
+ void testBuildCatalogStoreFactoryContextWithGenericInMemory() {
+ Configuration configuration = new Configuration();
+ configuration.set(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND, "generic_in_memory");
+ configuration.setString("table.catalog-store.generic_in_memory.option1", "value1");
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+
+ CatalogStoreFactory.Context context =
+ ApiFactoryUtil.buildCatalogStoreFactoryContext(configuration, classLoader);
+
+ assertThat(context).isNotNull();
+ assertThat(context.getOptions())
+ .containsExactlyInAnyOrderEntriesOf(Map.of("option1", "value1"));
+ assertThat(context.getConfiguration()).isEqualTo(configuration);
+ assertThat(context.getClassLoader()).isEqualTo(classLoader);
+ }
+
+ @Test
+ void testBuildCatalogStoreFactoryContextWithoutOptions() {
+ Configuration configuration = new Configuration();
+ configuration.set(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND, "generic_in_memory");
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+
+ CatalogStoreFactory.Context context =
+ ApiFactoryUtil.buildCatalogStoreFactoryContext(configuration, classLoader);
+
+ assertThat(context).isNotNull();
+ assertThat(context.getOptions()).isEmpty();
+ assertThat(context.getConfiguration()).isEqualTo(configuration);
+ assertThat(context.getClassLoader()).isEqualTo(classLoader);
+ }
+
+ @Test
+ void testBuildCatalogStoreFactoryContextOnlyExtractsRelevantOptions() {
+ Configuration configuration = new Configuration();
+ configuration.set(CommonCatalogOptions.TABLE_CATALOG_STORE_KIND, "file");
+ configuration.setString("table.catalog-store.file.path", "/test/path");
+ configuration.setString("table.catalog-store.file.option1", "value1");
+ configuration.setString("table.catalog-store.other.irrelevant", "should-not-appear");
+ configuration.setString("other.config.key", "should-not-appear");
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+
+ CatalogStoreFactory.Context context =
+ ApiFactoryUtil.buildCatalogStoreFactoryContext(configuration, classLoader);
+
+ assertThat(context).isNotNull();
+ assertThat(context.getOptions())
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of("path", "/test/path", "option1", "value1"));
+ }
+
+ private static Stream<Arguments> catalogStoreFactoryTestParameters() {
+ return Stream.of(
+ Arguments.of("generic_in_memory", GenericInMemoryCatalogStoreFactory.class),
+ Arguments.of("file", FileCatalogStoreFactory.class),
+ Arguments.of(null, GenericInMemoryCatalogStoreFactory.class));
+ }
+}
diff --git a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala
index 7e86c5bf256..c8cc4e99463 100644
--- a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala
+++ b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/internal/StreamTableEnvironmentImpl.scala
@@ -28,7 +28,7 @@ import org.apache.flink.table.catalog._
import org.apache.flink.table.connector.ChangelogMode
import org.apache.flink.table.delegation.{Executor, Planner}
import org.apache.flink.table.expressions.Expression
-import org.apache.flink.table.factories.{PlannerFactoryUtil, TableFactoryUtil}
+import org.apache.flink.table.factories.{ApiFactoryUtil, PlannerFactoryUtil, TableFactoryUtil}
import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, TableFunction, UserDefinedFunctionHelper}
import org.apache.flink.table.legacy.sources.TableSource
import org.apache.flink.table.module.ModuleManager
@@ -250,14 +250,19 @@ object StreamTableEnvironmentImpl {
val resourceManager = new ResourceManager(settings.getConfiguration, userClassLoader)
val moduleManager = new ModuleManager
- val catalogStoreFactory =
- TableFactoryUtil.findAndCreateCatalogStoreFactory(settings.getConfiguration, userClassLoader)
- val catalogStoreFactoryContext =
- TableFactoryUtil.buildCatalogStoreFactoryContext(settings.getConfiguration, userClassLoader)
- catalogStoreFactory.open(catalogStoreFactoryContext)
- val catalogStore =
- if (settings.getCatalogStore != null) settings.getCatalogStore
- else catalogStoreFactory.createCatalogStore()
+ val (catalogStore, catalogStoreFactory) =
+ if (settings.getCatalogStore.isPresent) {
+ (settings.getCatalogStore.get(), null)
+ } else {
+ val factory =
+ ApiFactoryUtil.findAndCreateCatalogStoreFactory(
+ settings.getConfiguration,
+ userClassLoader)
+ val factoryContext =
+ ApiFactoryUtil.buildCatalogStoreFactoryContext(settings.getConfiguration, userClassLoader)
+ factory.open(factoryContext)
+ (factory.createCatalogStore(), factory)
+ }
val catalogManager = CatalogManager.newBuilder
.classLoader(userClassLoader)