[fix][flink] Fix Iceberg compatible metadata cannot be read when data file format is parquet #4288

tsreaper · 2024-10-08T10:23:23Z

Purpose

Currently there is not IT cases for Paimon Iceberg compatibility in Flink. This PR adds IT cases and fixes some issues with parquet format.

Tests

IT cases.

API and Format

No format changes.

Documentation

Document for the whole Iceberg compatibility feature is also added.

JingsongLi · 2024-10-09T08:03:26Z

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java

+                    .defaultValue(50);
+
+    /** Where to store Iceberg metadata. */
+    public enum StorageType {


implements DescribedEnum

JingsongLi · 2024-10-09T08:04:21Z

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java

+                    .defaultValue(50);
+
+    /** Where to store Iceberg metadata. */
+    public enum StorageType {


Values can be: NONE, TABLE_LOCATION, HADOOP_CATALOG?

Why NONE? This value should not be set if users don't need Iceberg compatibility.

Or name it DISABLED

Generally speaking, using ENUM already has a state inside, and disable should also be one of its states

JingsongLi · 2024-10-09T10:12:15Z

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java

+        HADOOP_CATALOG(
+                "Store Iceberg metadata in a separate directory. "
+                        + "This directory can be specified as the warehouse directory of an Iceberg Hadoop catalog.");
+


You can add a value and implement toString. For the paimon style, the value should be table-location and hadoop-catalog.

wwj6591812 · 2024-10-09T16:55:24Z

paimon-core/src/main/java/org/apache/paimon/iceberg/AbstractIcebergCommitCallback.java

+                                    dbPath.getParent(),
+                                    "iceberg/"
+                                            + dbPath.getName()
+                                                    .substring(0, dbPath.getName().length() - 3)


1、private final String xxx = ".db";
2、Change "substring(0, dbPath.getName().length() - 3)" to ""substring(0, dbPath.getName().length() - xxx.length())"

wwj6591812 · 2024-10-09T16:57:23Z

paimon-core/src/main/java/org/apache/paimon/iceberg/AbstractIcebergCommitCallback.java

-    private List<IcebergPartitionField> getPartitionFields(RowType partitionType) {
+    private List<IcebergPartitionField> getPartitionFields(
+            List<String> partitionKeys, IcebergSchema icebergSchema) {
+        Map<String, IcebergDataField> fields = new HashMap<>();


Map<String, IcebergDataField> fields = new HashMap<>(icebergSchema.fields().size());

wwj6591812 · 2024-10-09T16:58:04Z

paimon-core/src/main/java/org/apache/paimon/iceberg/AbstractIcebergCommitCallback.java

+        for (IcebergDataField field : icebergSchema.fields()) {
+            fields.put(field.name(), field);
+        }
+
        List<IcebergPartitionField> result = new ArrayList<>();


List result = new ArrayList<>(partitionKeys.size());

wwj6591812 · 2024-10-09T17:00:17Z

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java

+    public static final ConfigOption<Integer> COMPACT_MIN_FILE_NUM =
+            ConfigOptions.key("metadata.iceberg.compaction.min.file-num")
+                    .intType()
+                    .defaultValue(10);


add withDescription?

wwj6591812 · 2024-10-09T17:00:25Z

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java

+    public static final ConfigOption<Integer> COMPACT_MAX_FILE_NUM =
+            ConfigOptions.key("metadata.iceberg.compaction.max.file-num")
+                    .intType()
+                    .defaultValue(50);


add withDescription?

wwj6591812 · 2024-10-09T17:04:10Z

paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergSchema.java

-        this(
+    public static IcebergSchema create(TableSchema tableSchema) {
+        int bias;
+        if (new CoreOptions(tableSchema.options()).formatType().equals("parquet")) {


use CoreOptions.FILE_FORMAT_PARQUET

wwj6591812 · 2024-10-09T17:07:05Z

paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergSchema.java

                (int) tableSchema.id(),
                tableSchema.fields().stream()
-                        .map(IcebergDataField::new)
+                        .map(f -> new IcebergDataField(f, bias))
                        .collect(Collectors.toList()));
    }


add "private static final String STRUCT = "struct";" for line 77

wwj6591812 · 2024-10-09T17:08:59Z

paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java

@@ -165,7 +166,8 @@ protected List<CommitCallback> createCommitCallbacks(String commitUser) {
        List<CommitCallback> callbacks = super.createCommitCallbacks(commitUser);
        CoreOptions options = coreOptions();

-        if (options.metadataIcebergCompatible()) {
+        if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)


Move this to AbstractFileStoreTable for AppendOnlyFileStoreTable and PrimaryKeyFileStoreTable?

wwj6591812 · 2024-10-09T17:09:05Z

paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java

@@ -185,7 +186,8 @@ protected List<CommitCallback> createCommitCallbacks(String commitUser) {
        List<CommitCallback> callbacks = super.createCommitCallbacks(commitUser);
        CoreOptions options = coreOptions();

-        if (options.metadataIcebergCompatible()) {
+        if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
+                != IcebergOptions.StorageType.DISABLED) {


Move this to AbstractFileStoreTable for AppendOnlyFileStoreTable and PrimaryKeyFileStoreTable?

wwj6591812 · 2024-10-09T17:11:42Z

...aimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java

+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for Paimon Iceberg compatibility. */


Base IT cases for Paimon Iceberg compatibility.

…docs.

JingsongLi

Looks good to me!

tsreaper added 2 commits October 8, 2024 18:22

[flink] Add IT cases for Iceberg compatibility

e350080

[fix] Fix failed tests

6ba88f1

tsreaper changed the title ~~[flink] Add IT cases for Iceberg compatibility~~ [fix][flink] Fix Iceberg compatible metadata cannot be read when data file format is parquet Oct 9, 2024

JingsongLi reviewed Oct 9, 2024

View reviewed changes

[fix] Fix comments

4cfb656

JingsongLi reviewed Oct 9, 2024

View reviewed changes

[fix] Add append-only table tests

e0a07a4

wwj6591812 reviewed Oct 9, 2024

View reviewed changes

[fix] Fix comments

df51b5a

tsreaper marked this pull request as draft October 10, 2024 04:34

tsreaper added 3 commits October 10, 2024 13:29

[fix] Add filter tests

8181557

[fix] Fix comments

55a2da3

[fix] Fix failed tests

d833339

tsreaper force-pushed the iceberg-test branch from 3e7a17f to d833339 Compare October 10, 2024 07:13

[fix] Clean metadata after table is dropped and re-created. Also add …

af7b72e

…docs.

tsreaper marked this pull request as ready for review October 10, 2024 11:22

JingsongLi approved these changes Oct 11, 2024

View reviewed changes

JingsongLi merged commit 524915e into apache:master Oct 11, 2024
11 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][flink] Fix Iceberg compatible metadata cannot be read when data file format is parquet #4288

[fix][flink] Fix Iceberg compatible metadata cannot be read when data file format is parquet #4288

tsreaper commented Oct 8, 2024 •

edited

Loading

JingsongLi Oct 9, 2024 •

edited

Loading

JingsongLi Oct 9, 2024 •

edited

Loading

tsreaper Oct 9, 2024

JingsongLi Oct 9, 2024

JingsongLi Oct 9, 2024

JingsongLi Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

wwj6591812 Oct 9, 2024

JingsongLi left a comment


		import static org.assertj.core.api.Assertions.assertThat;

		/** IT cases for Paimon Iceberg compatibility. */

[fix][flink] Fix Iceberg compatible metadata cannot be read when data file format is parquet #4288

[fix][flink] Fix Iceberg compatible metadata cannot be read when data file format is parquet #4288

Conversation

tsreaper commented Oct 8, 2024 • edited Loading

Purpose

Tests

API and Format

Documentation

JingsongLi Oct 9, 2024 • edited Loading

Choose a reason for hiding this comment

JingsongLi Oct 9, 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

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

JingsongLi left a comment

Choose a reason for hiding this comment

tsreaper commented Oct 8, 2024 •

edited

Loading

JingsongLi Oct 9, 2024 •

edited

Loading

JingsongLi Oct 9, 2024 •

edited

Loading