[#3515] feat(flink-connector): Support flink iceberg catalog #5914

sunxiaojian · 2024-12-19T07:05:30Z

What changes were proposed in this pull request?

Support flink iceberg catalog

Why are the changes needed?

Fix: #3515

Does this PR introduce any user-facing change?

no

How was this patch tested?

FlinkIcebergCatalogIT
FlinkIcebergHiveCatalogIT

sunxiaojian · 2024-12-19T09:57:27Z

@FANNG1 @coolderli PTAL

FANNG1 · 2024-12-19T11:05:45Z

Cool!, I'll review this PR, but may need some time, :)

sunxiaojian · 2024-12-19T11:28:37Z

Cool!, I'll review this PR, but may need some time, :)

ok, thanks

tengqm · 2024-12-22T04:24:42Z

docs/flink-connector/flink-catalog-iceberg.md

+The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.
+


Suggested change

The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.

The Apache Gravitino Flink connector can be used to read and write Iceberg tables, with the metadata managed by the Gravitino server.

To enable the Flink connector, you must download the Iceberg Flink runtime JAR and place it in the Flink classpath.

tengqm · 2024-12-22T04:25:13Z

docs/flink-connector/flink-catalog-iceberg.md

+- `INSERT INTO & OVERWRITE`
+- `SELECT`
+
+#### Not supported operations:


Suggested change

#### Not supported operations:

#### Operations Not Supported:

tengqm · 2024-12-22T04:26:49Z

docs/flink-connector/flink-catalog-iceberg.md

+
+## Catalog properties
+
+Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.


Suggested change

Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.

The Gravitino Flink connector transforms the following properties in a catalog to Flink connector configuration.

tengqm · 2024-12-22T04:28:02Z

docs/flink-connector/flink-catalog-iceberg.md

+
+### S3
+
+You need to add s3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`. Additionally, download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) and place it in the classpath of Flink.


Suggested change

You need to add s3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`. Additionally, download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) and place it in the classpath of Flink.

You need to add an S3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`.

Additionally, you need to download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle)

and place it in the Flink classpath.

tengqm · 2024-12-22T04:28:48Z

docs/flink-connector/flink-catalog-iceberg.md

+
+### OSS
+
+You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.


Suggested change

You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.

You need to add an OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`.

Additionally, you need download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip),

and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` to the Flink classpath.

tengqm · 2024-12-22T04:29:49Z

docs/flink-connector/flink-catalog-iceberg.md

+
+### GCS
+
+No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.


Suggested change

No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.

No extra configuration is needed. Please make sure the credential file is accessible by Flink.

For example, `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`.

You need to download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it in the Flink classpath.

tengqm · 2024-12-22T04:33:23Z

...c/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java

+        FactoryUtils.createCatalogFactoryHelper(this, context);
+    return new GravitinoIcebergCatalog(
+        context.getName(),
+        helper.getOptions().get(GravitinoIcebergCatalogFactoryOptions.DEFAULT_DATABASE),


Do we have to use such a long name?
Maybe CatalogFactoryOptions or even FactoryOptions is enough in such a well-defined package hierarchy?

Not recommended. If multiple CatalogFactoryOptions are referenced within the same class, the package name may have to be included every time

...c/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java

tengqm · 2024-12-22T04:35:43Z

...k/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConstants.java

+
+  String GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE = "hive";
+
+  @VisibleForTesting String ICEBERG_CATALOG_BACKEND_REST = CatalogUtil.ICEBERG_CATALOG_TYPE_REST;


Most, if not all fields in this interface are constants, right?

Yes, they are all constants

...k/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConverter.java

docs/flink-connector/flink-catalog-iceberg.md

FANNG1 · 2024-12-26T13:57:17Z

Hi, @sunxiaojian , Sorry for the delay, I'm working on the issues to release 0.8, may doesn't have enough time to review this PR these days.

xunliu · 2024-12-28T06:49:05Z

hi @sunxiaojian Thank you for your contributions.
Can you send an email to me([email protected]), I have something need to discuss with you.

sunxiaojian · 2024-12-29T14:32:38Z

Hi, @sunxiaojian , Sorry for the delay, I'm working on the issues to release 0.8, may doesn't have enough time to review this PR these days.

@FANNG1 ok , I will also handle the comments above as soon as possible

sunxiaojian · 2024-12-29T14:40:46Z

hi @sunxiaojian Thank you for your contributions. Can you send an email to me([email protected]), I have something need to discuss with you.

@xunliu The email has been sent

FANNG1 · 2025-01-10T09:37:26Z

docs/flink-connector/flink-catalog-iceberg.md

+- Metadata tables, like:
+  - `{iceberg_catalog}.{iceberg_database}.{iceberg_table}&snapshots`
+- Querying UDF
+  - `UPDATE` clause


UPDATE & DELETE clause are not under Query UDF?

FANNG1 · 2025-01-10T09:39:36Z

docs/flink-connector/flink-catalog-iceberg.md

+| `uri`                           | `uri`                                 | Catalog backend URI                                                                                                                                                                                                 | 0.8.0-incubating  |
+| `warehouse`                     | `warehouse`                           | Catalog backend warehouse                                                                                                                                                                                           | 0.8.0-incubating  |
+| `io-impl`                       | `io-impl`                             | The IO implementation for `FileIO` in Iceberg.                                                                                                                                                                      | 0.8.0-incubating  |
+| `oss-endpoint`                  | `oss.endpoint`                        | The endpoint of Aliyun OSS service.                                                                                                                                                                                 | 0.8.0-incubating  |


in 0.8, Gravitino relax the constrait for the secret key to pass the client, could you add OSS AKSK too?

FANNG1 · 2025-01-10T09:43:48Z

...k/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConverter.java

+
+  private IcebergPropertiesConverter() {}
+
+  private static final Map<String, String> GRAVITINO_CONFIG_TO_FLINK_ICEBERG;


why not initiate the map directly?

FANNG1 · 2025-01-10T09:46:47Z

...k/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConverter.java

+    all.put(
+        CommonCatalogOptions.CATALOG_TYPE.key(), GravitinoIcebergCatalogFactoryOptions.IDENTIFIER);
+    // Map "catalog-backend" to "catalog-type".
+    GRAVITINO_CONFIG_TO_FLINK_ICEBERG.forEach(


Gravitino supports custom catalog backend, should we do special logic here?

FANNG1 · 2025-01-10T09:50:43Z

...org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergHiveCatalogIT.java

+import org.junit.jupiter.api.Tag;
+
+@Tag("gravitino-docker-test")
+public class FlinkIcebergHiveCatalogIT extends FlinkIcebergCatalogIT {


Does it support other catalog backend like JDBC or REST?

It should be supported. This time, I only added Hive tests. In the future, I will add unit tests for JDBC and REST. There may be some minor adjustments in the logic, so detailed unit tests are necessary.

ok, could you add the supported catalog backend to the document? we could update the document when we add tests for other catalog backend.

sunxiaojian changed the title ~~[#3515]feat(flink-connector)Support flink iceberg catalog~~ [#3515] feat(flink-connector): Support flink iceberg catalog Dec 19, 2024

sunxiaojian force-pushed the support-flink-iceberg-catalog branch 4 times, most recently from 24cd6b8 to ef294ba Compare December 19, 2024 07:37

tengqm reviewed Dec 22, 2024

View reviewed changes

coolderli reviewed Dec 23, 2024

View reviewed changes

...k/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConverter.java Outdated Show resolved Hide resolved

FANNG1 reviewed Dec 26, 2024

View reviewed changes

docs/flink-connector/flink-catalog-iceberg.md Outdated Show resolved Hide resolved

sunxiaojian force-pushed the support-flink-iceberg-catalog branch 3 times, most recently from 4314cc7 to fc59502 Compare January 3, 2025 17:53

Support flink iceberg catalog apache#3515

5acadb8

sunxiaojian force-pushed the support-flink-iceberg-catalog branch from fc59502 to 5acadb8 Compare January 4, 2025 03:33

FANNG1 reviewed Jan 10, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

sunxiaojian commented Dec 19, 2024

sunxiaojian commented Dec 19, 2024

FANNG1 commented Dec 19, 2024

sunxiaojian commented Dec 19, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

tengqm Dec 22, 2024

sunxiaojian Dec 30, 2024

tengqm Dec 22, 2024

sunxiaojian Dec 22, 2024

FANNG1 commented Dec 26, 2024

xunliu commented Dec 28, 2024

sunxiaojian commented Dec 29, 2024

sunxiaojian commented Dec 29, 2024 •

edited

Loading

FANNG1 Jan 10, 2025

FANNG1 Jan 10, 2025

FANNG1 Jan 10, 2025

FANNG1 Jan 10, 2025

FANNG1 Jan 10, 2025

sunxiaojian Jan 11, 2025

FANNG1 Jan 13, 2025

		The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.

-The Apache Gravitino Flink connector offers the capability to read and write Iceberg tables, with the metadata managed by the Gravitino server. To enable the use of the Iceberg catalog within the Flink connector, you must download the Iceberg Flink runtime JAR to the Flink classpath.
+The Apache Gravitino Flink connector can be used to read and write Iceberg tables, with the metadata managed by the Gravitino server.
+To enable the Flink connector, you must download the Iceberg Flink runtime JAR and place it in the Flink classpath.

	#### Not supported operations:
	#### Operations Not Supported:


		## Catalog properties

		Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.

	Gravitino Flink connector will transform the following property names defined in catalog properties to Flink Iceberg connector configuration.
	The Gravitino Flink connector transforms the following properties in a catalog to Flink connector configuration.


		### S3

		You need to add s3 secret to the Flink configuration using `s3.access-key-id` and `s3.secret-access-key`. Additionally, download the [Iceberg AWS bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws-bundle) and place it in the classpath of Flink.


		### OSS

		You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.

-You need to add OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`. Additionally, download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip) and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` in the classpath of Flink.
+You need to add an OSS secret key to the Flink configuration using `client.access-key-id` and `client.access-key-secret`.
+Additionally, you need download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip),
+and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` to the Flink classpath.


		### GCS

		No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.

-No extra configuration is needed. Please make sure the credential file is accessible by Flink, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`, and download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it to the classpath of Flink.
+No extra configuration is needed. Please make sure the credential file is accessible by Flink.
+For example, `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`.
+You need to download [Iceberg GCP bundle](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-gcp-bundle) and place it in the Flink classpath.


		String GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE = "hive";

		@VisibleForTesting String ICEBERG_CATALOG_BACKEND_REST = CatalogUtil.ICEBERG_CATALOG_TYPE_REST;


		private IcebergPropertiesConverter() {}

		private static final Map<String, String> GRAVITINO_CONFIG_TO_FLINK_ICEBERG;

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

Are you sure you want to change the base?

[#3515] feat(flink-connector): Support flink iceberg catalog #5914

Conversation

sunxiaojian commented Dec 19, 2024

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?

sunxiaojian commented Dec 19, 2024

FANNG1 commented Dec 19, 2024

sunxiaojian commented Dec 19, 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

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 Dec 26, 2024

xunliu commented Dec 28, 2024

sunxiaojian commented Dec 29, 2024

sunxiaojian commented Dec 29, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sunxiaojian commented Dec 29, 2024 •

edited

Loading