[#3362] feat(flink-connector): Add the code skeleton for flink-connector #2635

coolderli · 2024-03-22T06:47:25Z

What changes were proposed in this pull request?

support GravitinoCatalogStore to register the catalog. In the MR, we will support to create the hive catalog.

Why are the changes needed?

Fix [Subtask] [flink-connector] Support GravitinoCatalogStore to register the catalog #3362

Does this PR introduce any user-facing change?

support flink in gravitino

How was this patch tested?

add UTs

jerryshao · 2024-05-13T06:58:36Z

@coolderli can we please move forward the flink support?

coolderli · 2024-05-13T07:41:31Z

@coolderli can we please move forward the flink support?

@jerryshao Of course. I will finish the first one today.

coolderli · 2024-05-13T10:49:08Z

@jerryshao @FANNG1 Could you please help view this? Thanks.

coolderli · 2024-05-13T11:09:04Z

...src/main/java/com/datastrato/gravitino/flink/connector/hive/GravitinoHiveCatalogFactory.java

+
+  @Override
+  public Catalog createCatalog(Context context) {
+    this.hiveCatalogFactory = new HiveCatalogFactory();


We use the HiveCatalogFactory from flink-connector-hive to create the HiveCatalog.

flink-connector/build.gradle.kts

integration-test/build.gradle.kts

...java/com/datastrato/gravitino/flink/connector/store/GravitinoCatalogStoreFactoryOptions.java

.../src/main/java/com/datastrato/gravitino/flink/connector/catalog/GravitinoCatalogManager.java

integration-test/build.gradle.kts

...ctor/src/main/java/com/datastrato/gravitino/flink/connector/store/GravitinoCatalogStore.java

FANNG1 · 2024-05-14T02:16:14Z

flink-connector/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory

+#
+
+com.datastrato.gravitino.flink.connector.store.GravitinoCatalogStoreFactory
+com.datastrato.gravitino.flink.connector.hive.GravitinoHiveCatalogFactory


If supporting Iceberg, just provide another GravitinoIcebergCatalogFactory ?

@FANNG1 Yes. This is what I want to talk about. Do we implement one CatalogFactory for each storage, or just one GravitinoCatalogFactory? If we use one GravitinoCatalogFactory, we may need another property to identify which real catalog should be used.

Another question is do we need to use GravitinoCatalog for all storage. I think the CatalogFactory should be consistent with the Catalog. Flink supports registering the catalog manually, if we only give a GravitinoCatalog, we can also simplify the usage.

The original usage:https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/catalogs/#using-sql-ddl

// Create a HiveCatalog Catalog catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>"); // Register the catalog tableEnv.registerCatalog("myhive", catalog);

Use one GravitinoCatalog.

Map<String, String> properties = new HashMap<>(); **properties.put("catalog.type", "hive"); // another property used to identify the real catalog** Catalog catalog = new GravitinoCatalog("gravitino", properties) // Register the catalog tableEnv.registerCatalog("myhive", catalog);

Use different GravitinoCatalog for different storage:

Catalog catalog = new GravitinoHiveCatalog("gravitino", xxx) // Register the catalog tableEnv.registerCatalog("myhive", catalog);

which do you prefer? Can you share your thoughts? Thanks.

Can you also help review it? Thanks. @hackergin

I prefer different GravitinoCatalog, because this solution seems simple for user.

From my understanding, the Catalog Store is essentially for storing the configuration of Flink Catalog. Therefore, if users create a catalog using the following DDL, we should support saving the configuration of the Catalog.

CREATE CATALOG myhive WITH ( 'type' = 'hive', 'hive-conf-dir' = '/opt/hive-conf' );

At the same time, when constructing a real Catalog instance, we can automatically create a GravitinoCatalog instead of a HiveCatalog.

And we should prohibit the direct creation of Catalog instances.

tableEnv.registerCatalog("myhive", catalog);

When using the Catalog instance to register this Catalog with the table env, this Catalog will not be persisted to the Catalog Store.

...nector/src/main/java/com/datastrato/gravitino/flink/connector/hive/GravitinoHiveCatalog.java

...src/main/java/com/datastrato/gravitino/flink/connector/hive/GravitinoHiveCatalogFactory.java

...c/main/java/com/datastrato/gravitino/flink/connector/store/GravitinoCatalogStoreFactory.java

coolderli · 2024-05-16T08:46:23Z

@FANNG1 I have updated the MR. Please help review it when you have time. Thanks.

flink-connector/build.gradle.kts

...ctor/src/main/java/com/datastrato/gravitino/flink/connector/store/GravitinoCatalogStore.java

...ctor/src/test/java/com/datastrato/gravitino/flink/connector/integration/test/FlinkEnvIT.java

flink-connector/src/test/resources/flink-tests/hive-site.xml

.../java/com/datastrato/gravitino/flink/connector/integration/test/hive/HiveCatalogStoreIT.java

...tor/src/main/java/com/datastrato/gravitino/flink/connector/hive/HivePropertiesConverter.java

coolderli · 2024-05-20T09:48:47Z

@FANNG1 Could you help review this MR again? Thanks.

coolderli · 2024-05-20T10:14:37Z

...src/main/java/com/datastrato/gravitino/flink/connector/hive/GravitinoHiveCatalogFactory.java

+    this.hiveCatalogFactory = new HiveCatalogFactory();
+    final FactoryUtil.CatalogFactoryHelper helper =
+        FactoryUtil.createCatalogFactoryHelper(this, context);
+    helper.validateExcept(


This will skip the validation with the specified prefix.

What if user add a properties like flink.bypass.aa by mistake, this will cause all Flink jobs fail? I prefer to ignore it.

@FANNG1 It's good for users. How about just removing the validation? Otherwise, I have to remove these useless key before the validation.

@FANNG1 I thought about it again. The validation is needed, it can be used to validate the required options such as metastore.uris. Let me check how to finish it.

please add comment before helper.validateExcept to descript how it works

...src/main/java/com/datastrato/gravitino/flink/connector/hive/GravitinoHiveCatalogFactory.java

coolderli · 2024-05-20T10:24:58Z

@FANNG1 I think I have already rebased the main branch. But the GitHub CI still failed. Can you help me with it? Thanks.

coolderli

@FANNG1 I have finished it. Make hive.metastore.uris as required and ignore the unknown key. Please take a look. Thanks.

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/utils/FactoryUtils.java

...n/java/com/datastrato/gravitino/flink/connector/hive/GravitinoHiveCatalogFactoryOptions.java

...src/main/java/com/datastrato/gravitino/flink/connector/hive/GravitinoHiveCatalogFactory.java

coolderli · 2024-05-23T12:17:53Z

.../java/com/datastrato/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java

+    Assertions.assertEquals(
+        "unknown.value",
+        flinkProperties.get(flinkByPass("unknown.key")),
+        "The unknown.key will not cause failure and will be saved in Gravitino.");


The unknown key will be saved in Gravitino but not be ignored.

Why does it works? the unknow.key not exists in requiredOptions or optionalOptions and the customer CatalogFactoryHelper only works when loading catalog from Gravitino. Do I miss something?

@FANNG1 When creating the catalog, it will pass the unknow.key to the Context of GravitinoHiveCatalog and the CatalogDescription of the GravitinoCatalogStore. The GravitinoCatalogStore will not validate the unknow.key. And we use FactoryUtils.GravitinoCatalogFactoryHelper instead of FactoryUtil.CatalogFactoryHelper to skip the validation. So the unknow.key will save to the Gravitino successfully.

Of course, there is a little weird about the key that is not in the optionalOptions not failing the Flink Job.

@FANNG1 Do you think this is reasonable?

it is a little weird,but let's go ahead

FANNG1 · 2024-05-24T04:01:48Z

@coolderli , thanks for proposing the base PR for Flink, a big step!

…connector (apache#2635)  ### What changes were proposed in this pull request? - support GravitinoCatalogStore to register the catalog. In the MR, we will support to create the hive catalog. ### Why are the changes needed? - Fix apache#3362 ### Does this PR introduce _any_ user-facing change? - support flink in gravitino ### How was this patch tested? - add UTs

coolderli marked this pull request as draft March 22, 2024 06:47

coolderli mentioned this pull request Mar 27, 2024

[EPIC] Add Flink engine support for Gravitino #1354

Open

10 tasks

coolderli force-pushed the init-flink branch from 3d9b3f9 to 543cc10 Compare April 21, 2024 12:10

coolderli force-pushed the init-flink branch 2 times, most recently from 6b4dff5 to d37b64b Compare May 6, 2024 03:51

coolderli force-pushed the init-flink branch from d37b64b to a71b72b Compare May 8, 2024 08:41

coolderli changed the title ~~feat(flink): support flink in gravitino~~ [#3362] feat(flink): support GravitinoCatalogStore to register the catalog May 13, 2024

coolderli marked this pull request as ready for review May 13, 2024 07:31

coolderli force-pushed the init-flink branch from a3aaa56 to c63d5ad Compare May 13, 2024 08:15

coolderli closed this May 13, 2024

coolderli reopened this May 13, 2024

coolderli commented May 13, 2024

View reviewed changes

FANNG1 reviewed May 13, 2024

View reviewed changes

flink-connector/build.gradle.kts Outdated Show resolved Hide resolved

integration-test/build.gradle.kts Outdated Show resolved Hide resolved

FANNG1 reviewed May 14, 2024

View reviewed changes

coolderli force-pushed the init-flink branch from 143e2c3 to d64235e Compare May 16, 2024 08:40

coolderli changed the title ~~[#3362] feat(flink): support GravitinoCatalogStore to register the catalog~~ [#3362] feat(flink-connector): support GravitinoCatalogStore to register the catalog May 16, 2024

FANNG1 reviewed May 17, 2024

View reviewed changes

...tor/src/main/java/com/datastrato/gravitino/flink/connector/hive/HivePropertiesConverter.java Outdated Show resolved Hide resolved

FANNG1 reviewed May 17, 2024

View reviewed changes

...tor/src/main/java/com/datastrato/gravitino/flink/connector/hive/HivePropertiesConverter.java Show resolved Hide resolved

FANNG1 reviewed May 17, 2024

View reviewed changes

...tor/src/main/java/com/datastrato/gravitino/flink/connector/hive/HivePropertiesConverter.java Show resolved Hide resolved

coolderli force-pushed the init-flink branch 2 times, most recently from 77806f4 to 7e20a11 Compare May 20, 2024 08:49

coolderli commented May 20, 2024

View reviewed changes

coolderli added 21 commits May 23, 2024 20:08

fix spotless

2811e2f

fix spotless

ce86208

fix it

df8492e

address comments

4e94a54

fix useless

a847a34

remove cache

63805a3

fix

7ea0687

test

356e399

test

c302c3e

debug

7931498

debug

f27ab23

add

91a0641

add

331fed5

add seperate github action

a1ed7be

fix

5ca92e4

fix

40a29f7

fix its

b8b047f

address comment

71dd28a

address comments

46f0e7d

fix uts

5b6f6da

ignore some options

7c3c342

coolderli force-pushed the init-flink branch from 09d12c9 to 7c3c342 Compare May 23, 2024 12:08

move the identifier

b546a45

coolderli commented May 23, 2024

View reviewed changes

coolderli added 2 commits May 23, 2024 20:21

fix rat

6f1cfaa

add some description

75b8a16

FANNG1 approved these changes May 24, 2024

View reviewed changes

FANNG1 merged commit 2113efd into apache:main May 24, 2024
26 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#3362] feat(flink-connector): Add the code skeleton for flink-connector #2635

[#3362] feat(flink-connector): Add the code skeleton for flink-connector #2635

coolderli commented Mar 22, 2024 •

edited

Loading

jerryshao commented May 13, 2024

coolderli commented May 13, 2024

coolderli commented May 13, 2024

coolderli May 13, 2024

FANNG1 May 14, 2024

coolderli May 14, 2024 •

edited

Loading

coolderli May 14, 2024

FANNG1 May 14, 2024

hackergin May 14, 2024

coolderli commented May 16, 2024

coolderli commented May 20, 2024

coolderli May 20, 2024

FANNG1 May 23, 2024

coolderli May 23, 2024

coolderli May 23, 2024

FANNG1 May 24, 2024

coolderli commented May 20, 2024

coolderli left a comment

coolderli May 23, 2024

FANNG1 May 24, 2024

coolderli May 24, 2024

coolderli May 24, 2024

FANNG1 May 24, 2024

FANNG1 commented May 24, 2024

[#3362] feat(flink-connector): Add the code skeleton for flink-connector #2635

[#3362] feat(flink-connector): Add the code skeleton for flink-connector #2635

Conversation

coolderli commented Mar 22, 2024 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

jerryshao commented May 13, 2024

coolderli commented May 13, 2024

coolderli commented May 13, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coolderli May 14, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coolderli commented May 16, 2024

coolderli commented May 20, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coolderli commented May 20, 2024

coolderli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FANNG1 commented May 24, 2024

coolderli commented Mar 22, 2024 •

edited

Loading

coolderli May 14, 2024 •

edited

Loading