pass catalog sensitive info when create fluss catalog #1937

xx789633 · 2025-11-05T10:06:39Z

Purpose

Linked issue: close #1903

Brief change log

Tests

API and Format

Documentation

luoyuxia

@xx789633 Thanks for the pr. But it should allow pass all options instead of sensitive table options only. I image some thing like:
create catalog t1 with (
'paimon.k1' = 'v1',
'paimon.k2 = 'v2'
)
to enable pass 'k1' = 'v1' as well as 'k2 = 'v2

xx789633 · 2025-11-10T15:27:25Z

You are correct @luoyuxia . Let me rename it.

LiebingYu · 2025-11-13T11:20:45Z

...ink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalogFactory.java

+        helper.validateExcept(
+                Stream.concat(
+                                Stream.of(CLIENT_SECURITY_PREFIX),
+                                Arrays.stream(DataLakeFormat.values())
+                                        .map(DataLakeFormat::toString))
+                        .toArray(String[]::new));


I suggest to make it extensiable. Thus if the subclass want to support new configs, they don't need to overwrite createCatalog.

protected static final List<String> ALLOW_CATALOG_CONFIG_PREFIX = new ArrayList<>(); static { ALLOW_CATALOG_CONFIG_PREFIX.add(CLIENT_SECURITY_PREFIX); // support lake catalog options for (DataLakeFormat value : DataLakeFormat.values()) { ALLOW_CATALOG_CONFIG_PREFIX.add(value.toString()); } }

Good suggestion. Done.

luoyuxia

@xx789633 Thanks for the pr. Left minor comments. PTAL
Also, is it possible to document this behavior that users can pass lake related properties to Flink Catalog via SQL DDL?

luoyuxia · 2025-12-15T09:49:22Z

fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java

                Collections.singletonList(ConfigOptions.TABLE_DATALAKE_ENABLED.key());
    }

+    public static final Set<String> SENSITIVE_TABLE_OPTIONS = new HashSet<>();


why move to here?

it seems similar configs like ALTERABLE_TABLE_OPTIONS are all in this file.

But currently, I don't want to expose them in FlussConfigUtils...

luoyuxia · 2025-12-15T09:50:32Z

fluss-common/src/main/java/org/apache/fluss/utils/PropertiesUtils.java

                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

+    public static <V> Map<String, V> extractPrefix(


why introducr a new method? Can't we reuse extractPrefix(Map<String, V> originalMap, String prefix)?

I suggest to handle in FlinkCatalogFactory. The method is hard to reuse...

done. just reuse extractPrefix(Map<String, V> originalMap, String prefix).

luoyuxia · 2025-12-15T10:01:35Z

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeFlinkCatalog.java

                                        + "' is set.");
                    }
+                    String dataLakePrefix = lakeFormat.toString() + ".";
+                    Map<String, String> catalogProperties =


Can be

Map<String, String> catalogProperties = PropertiesUtils.extractAndRemovePrefix( lakeCatalogProperties, lakeFormat + ".");

?

luoyuxia · 2025-12-15T10:04:35Z

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

        securityMap.put("client.security.sasl.username", "root");
        securityMap.put("client.security.sasl.password", "password");

+        Map<String, String> lakeCatalogMap = new HashMap<>();


add test in FlinkCatalogITCase to verify SQL can pass the properties.

a new test testCreateCatalogWithLakeProperties is added.

luoyuxia · 2025-12-15T11:10:23Z

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

    protected final String bootstrapServers;
    protected final Map<String, String> securityConfigs;
    protected final LakeFlinkCatalog lakeFlinkCatalog;
+    protected final Map<String, String> lakeCatalogProperties;


Could you please make lakeCatalogProperties to be Supplier<Map<String, String>> lakeCatalogPropertiesProvider. The reason it that it make it more plugable. In our inner use, we will retrieve the lake catalog related from flink conf, but we only want to do the retrieve if it's required.

make sense. done.

xx789633 · 2025-12-16T09:56:21Z

doc is updated. please take a look @luoyuxia

luoyuxia

@xx789633 Left some comments again

luoyuxia · 2025-12-16T13:21:59Z

fluss-flink/fluss-flink-2.1/src/main/java/org/apache/fluss/flink/catalog/Flink21Catalog.java

-            Map<String, String> securityConfigs) {
-        super(name, defaultDatabase, bootstrapServers, classLoader, securityConfigs);
+            Map<String, String> securityConfigs,
+            Supplier<Map<String, String>> lakeCatalogProperties) {


Suggested change

Supplier<Map<String, String>> lakeCatalogProperties) {

Supplier<Map<String, String>> lakeCatalogPropertiesSupplier) {

luoyuxia · 2025-12-16T13:22:08Z

fluss-flink/fluss-flink-2.1/src/main/java/org/apache/fluss/flink/catalog/Flink21Catalog.java

            String bootstrapServers,
            ClassLoader classLoader,
            Map<String, String> securityConfigs,
+            Supplier<Map<String, String>> lakeCatalogProperties,


luoyuxia · 2025-12-16T13:22:26Z

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

    protected final String bootstrapServers;
    protected final Map<String, String> securityConfigs;
    protected final LakeFlinkCatalog lakeFlinkCatalog;
+    protected final Supplier<Map<String, String>> lakeCatalogProperties;


luoyuxia · 2025-12-16T13:22:49Z

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

+                        objectPath.getDatabaseName(),
+                        tableName,
+                        tableInfo.getProperties(),
+                        lakeCatalogProperties);


nit: we can get in here directly
lakeCatalogProperties.get()

Acquiring the properties here will frequently invoke the supplier function and bring some overhead. Actually, the properties are only needed when creating the lake catalog.

luoyuxia · 2025-12-16T13:23:07Z

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

    }
+
+    @VisibleForTesting
+    public Supplier<Map<String, String>> getLakeCatalogProperties() {


Suggested change

public Supplier<Map<String, String>> getLakeCatalogProperties() {

public Map<String, String> getLakeCatalogProperties() {

…nk catalog (apache#1937)" This reverts commit 35d92c5.

…og (apache#1937)

pass catalog sensitive info when create fluss catalog

34c93df

luoyuxia reviewed Nov 10, 2025

View reviewed changes

nit

75a034c

xx789633 added 9 commits November 11, 2025 15:09

rename

6e43bd8

rename

db330cb

wip: pass all types of lake formats

038e69e

refactor

2531fd8

rename

e8f3cf6

Merge branch 'apache:main' into lakesense

37db330

fix quickstart sql

6dd0607

refactor

544344f

fix compile

3f5f7d8

LiebingYu reviewed Nov 13, 2025

View reviewed changes

xx789633 and others added 4 commits November 14, 2025 10:18

isolate PREFIXES_TO_SKIP_VALIDATE

9291067

remove quick start

af27626

remove quick start

bf03205

Merge branch 'main' into lakesense

6598ddf

luoyuxia reviewed Dec 15, 2025

View reviewed changes

xx789633 and others added 6 commits December 15, 2025 21:44

remove reduant code

51760d9

supplier

715997d

fix

7de5bd6

add test

3b415a2

move

a6a1fb9

move

c14537a

xx789633 added 2 commits December 16, 2025 20:24

Merge branch 'main' into lakesense

a0f1591

rebase

18b017f

luoyuxia reviewed Dec 16, 2025

View reviewed changes

rename

32826f5

fix supplier

15de5fd

luoyuxia approved these changes Dec 17, 2025

View reviewed changes

luoyuxia merged commit 35d92c5 into apache:main Dec 17, 2025
6 checks passed

LiebingYu added a commit to LiebingYu/fluss that referenced this pull request Dec 17, 2025

Revert "[flink] Allow pass lake catalog property info when create fli…

933314b

…nk catalog (apache#1937)" This reverts commit 35d92c5.

LiebingYu added a commit to LiebingYu/fluss that referenced this pull request Dec 18, 2025

Revert "[flink] Allow pass lake catalog property info when create fli…

8369177

…nk catalog (apache#1937)" This reverts commit 35d92c5.

luoyuxia mentioned this pull request Dec 19, 2025

[hotfix] Should pass to lake properties to lake table #2205

Merged

vamossagar12 pushed a commit to vamossagar12/fluss that referenced this pull request Dec 20, 2025

[flink] Allow pass lake catalog property info when create flink catal…

7b5f809

…og (apache#1937)

	Supplier<Map<String, String>> lakeCatalogProperties) {
	Supplier<Map<String, String>> lakeCatalogPropertiesSupplier) {

	public Supplier<Map<String, String>> getLakeCatalogProperties() {
	public Map<String, String> getLakeCatalogProperties() {

pass catalog sensitive info when create fluss catalog #1937

pass catalog sensitive info when create fluss catalog #1937

Uh oh!

Conversation

xx789633 commented Nov 5, 2025

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

xx789633 commented Nov 10, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xx789633 commented Dec 16, 2025

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants