[Feature][Connectors-v2] Paimon version upgrade to 1.0.1

Co-authored-by: Jia Fan <[email protected]>

Author: hawk9821

docs/en/connector-v2/sink/Paimon.md

@@ -6,6 +6,15 @@
 
 Sink connector for Apache Paimon. It can support cdc mode 、auto create table.
 
+### Comparison between Seatunnel and Paimon vsrsion
+
+| Seatunnel Version | Paimon Version   |
+|-------------------|------------------|
+| 2.3.2  -  2.3.3   | 0.4-SNAPSHOT     |
+| 2.3.4             | 0.6-SNAPSHOT     |
+| 2.3.5  -  2.3.9   | 0.7.0-incubating |
+| 2.3.10            | 1.0.1            |
+
 ## Supported DataSource Info
 
 | Datasource | Dependent |                                   Maven                                   |

docs/en/connector-v2/source/Paimon.md

@@ -6,6 +6,15 @@
 
 Read data from Apache Paimon.
 
+### Comparison between Seatunnel and Paimon vsrsion
+
+| Seatunnel Version | Paimon Version   |
+|-------------------|------------------|
+| 2.3.2  -  2.3.3   | 0.4-SNAPSHOT     |
+| 2.3.4             | 0.6-SNAPSHOT     |
+| 2.3.5  -  2.3.9   | 0.7.0-incubating |
+| 2.3.10            | 1.0.1            |
+
 ## Key features
 
 - [x] [batch](../../concept/connector-v2-features.md)

docs/zh/connector-v2/sink/Paimon.md

@@ -6,6 +6,15 @@
 
 Apache Paimon数据连接器。支持cdc写以及自动建表。
 
+### Seatunnmel与Paimon版本对照
+
+| Seatunnel Version | Paimon Version   |
+|-------------------|------------------|
+| 2.3.2  -  2.3.3   | 0.4-SNAPSHOT     |
+| 2.3.4             | 0.6-SNAPSHOT     |
+| 2.3.5  -  2.3.9   | 0.7.0-incubating |
+| 2.3.10            | 1.0.1            |
+
 ## 支持的数据源信息
 
 |  数据源   |    依赖     |                                   Maven                                   |

seatunnel-connectors-v2/connector-paimon/pom.xml

@@ -30,7 +30,7 @@
     <name>SeaTunnel : Connectors V2 : Paimon</name>
 
     <properties>
-        <paimon.version>0.7.0-incubating</paimon.version>
+        <paimon.version>1.0.1</paimon.version>
         <hive.version>2.3.9</hive.version>
         <connector.name>connector.paimon</connector.name>
     </properties>

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalog.java

@@ -38,6 +38,7 @@
 import org.apache.seatunnel.connectors.seatunnel.paimon.sink.PaimonSink;
 import org.apache.seatunnel.connectors.seatunnel.paimon.utils.SchemaUtil;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
@@ -51,6 +52,8 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -103,7 +106,8 @@ public String getDefaultDatabase() throws CatalogException {
 
     @Override
     public boolean databaseExists(String databaseName) throws CatalogException {
-        return catalog.databaseExists(databaseName);
+        List<String> listDatabases = catalog.listDatabases();
+        return listDatabases.contains(databaseName);
     }
 
     @Override
@@ -123,7 +127,16 @@ public List<String> listTables(String databaseName)
 
     @Override
     public boolean tableExists(TablePath tablePath) throws CatalogException {
-        return catalog.tableExists(toIdentifier(tablePath));
+        Identifier identifier = toIdentifier(tablePath);
+        List<String> tables = new ArrayList<>();
+        try {
+            if (databaseExists(identifier.getDatabaseName())) {
+                tables = catalog.listTables(identifier.getDatabaseName());
+            }
+        } catch (org.apache.paimon.catalog.Catalog.DatabaseNotExistException e) {
+            return false;
+        }
+        return tables.contains(identifier.getTableName());
     }
 
     @Override
@@ -216,7 +229,9 @@ private Schema buildPaimonSchema(@NonNull org.apache.paimon.schema.TableSchema s
         Schema.Builder builder = Schema.newBuilder();
         schema.fields()
                 .forEach(field -> builder.column(field.name(), field.type(), field.description()));
-        builder.options(schema.options());
+        Map<String, String> options = new HashMap<>(schema.options());
+        options.remove(CoreOptions.PATH.key());
+        builder.options(options);
         builder.primaryKey(schema.primaryKeys());
         builder.partitionKeys(schema.partitionKeys());
         builder.comment(schema.comment());
@@ -296,7 +311,9 @@ private void resolveException(Exception e) {
             }
         } else if (cause instanceof RuntimeException) {
             String message = cause.getMessage();
-            if (message.contains("Cannot define 'bucket-key' in unaware or dynamic bucket mode.")) {
+            // https://github.com/apache/paimon/pull/3320/files#diff-d3e068ea8caf83d2371f0eaa1cbf3d02ff06e1c1cdceec5fab2e065cecd96230
+            if (message.contains(
+                    "Cannot define 'bucket-key' with bucket -1, please specify a bucket number.")) {
                 throw new PaimonConnectorException(
                         PaimonConnectorErrorCode.WRITE_PROPS_BUCKET_KEY_ERROR, message);
             }

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java

@@ -128,10 +128,9 @@ public PaimonSinkWriter(
         this.sinkPaimonTableSchema = this.paimonFileStoretable.schema();
         this.newTableWrite();
         BucketMode bucketMode = this.paimonFileStoretable.bucketMode();
-        this.dynamicBucket =
-                BucketMode.DYNAMIC == bucketMode || BucketMode.GLOBAL_DYNAMIC == bucketMode;
+        this.dynamicBucket = BucketMode.HASH_DYNAMIC == bucketMode;
         int bucket = ((FileStoreTable) paimonFileStoretable).coreOptions().bucket();
-        if (bucket == -1 && BucketMode.UNAWARE == bucketMode) {
+        if (bucket == -1 && BucketMode.BUCKET_UNAWARE == bucketMode) {
             log.warn("Append only table currently do not support dynamic bucket");
         }
         if (dynamicBucket) {

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/bucket/PaimonBucketAssigner.java

@@ -29,6 +29,8 @@
 import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Int2ShortHashMap;
 
 import java.io.IOException;
 import java.util.List;
@@ -37,6 +39,8 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static org.apache.paimon.crosspartition.IndexBootstrap.BUCKET_FIELD;
+
 public class PaimonBucketAssigner {
 
     private final RowPartitionKeyExtractor extractor;
@@ -45,31 +49,37 @@ public class PaimonBucketAssigner {
 
     private final TableSchema schema;
 
+    private final Int2ShortHashMap hash2Bucket;
+
     public PaimonBucketAssigner(Table table, int numAssigners, int assignId) {
         FileStoreTable fileStoreTable = (FileStoreTable) table;
         this.schema = fileStoreTable.schema();
         this.extractor = new RowPartitionKeyExtractor(fileStoreTable.schema());
-        long dynamicBucketTargetRowNum =
-                ((FileStoreTable) table).coreOptions().dynamicBucketTargetRowNum();
+        long dynamicBucketTargetRowNum = fileStoreTable.coreOptions().dynamicBucketTargetRowNum();
         this.simpleHashBucketAssigner =
                 new SimpleHashBucketAssigner(numAssigners, assignId, dynamicBucketTargetRowNum);
+        this.hash2Bucket = new Int2ShortHashMap();
         loadBucketIndex(fileStoreTable, numAssigners, assignId);
     }
 
     private void loadBucketIndex(FileStoreTable fileStoreTable, int numAssigners, int assignId) {
         IndexBootstrap indexBootstrap = new IndexBootstrap(fileStoreTable);
+        RowType indexRowType = IndexBootstrap.bootstrapType(schema);
+        int bucketFieldIndex = indexRowType.getFieldIndex(BUCKET_FIELD);
         List<String> fieldNames = schema.fieldNames();
         Map<String, Integer> fieldIndexMap =
                 IntStream.range(0, fieldNames.size())
                         .boxed()
                         .collect(Collectors.toMap(fieldNames::get, Function.identity()));
         List<DataField> primaryKeys = schema.primaryKeysFields();
+
         try (RecordReader<InternalRow> recordReader =
                 indexBootstrap.bootstrap(numAssigners, assignId)) {
             RecordReaderIterator<InternalRow> readerIterator =
                     new RecordReaderIterator<>(recordReader);
             while (readerIterator.hasNext()) {
                 InternalRow row = readerIterator.next();
+                int rowBucket = row.getInt(bucketFieldIndex);
                 GenericRow binaryRow = new GenericRow(fieldNames.size());
                 for (int i = 0; i < primaryKeys.size(); i++) {
                     String name = primaryKeys.get(i).name();
@@ -78,16 +88,19 @@ private void loadBucketIndex(FileStoreTable fileStoreTable, int numAssigners, in
                             fieldIndexMap.get(name),
                             InternalRow.createFieldGetter(type, i).getFieldOrNull(row));
                 }
-                assign(binaryRow);
+                hash2Bucket.put(
+                        extractor.trimmedPrimaryKey(binaryRow).hashCode(), (short) rowBucket);
             }
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
     }
 
     public int assign(InternalRow rowData) {
-        int hash = extractor.trimmedPrimaryKey(rowData).hashCode();
-        return Math.abs(
-                this.simpleHashBucketAssigner.assign(this.extractor.partition(rowData), hash));
+        int rowHashCode = extractor.trimmedPrimaryKey(rowData).hashCode();
+        if (hash2Bucket.containsKey(rowHashCode)) {
+            return hash2Bucket.get(rowHashCode);
+        }
+        return this.simpleHashBucketAssigner.assign(this.extractor.partition(rowData), rowHashCode);
     }
 }

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceReader.java

@@ -78,9 +78,9 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
             if (Objects.nonNull(split)) {
                 // read logic
                 try (final RecordReader<InternalRow> reader =
-                        tableRead.executeFilter().createReader(split.getSplit())) {
-                    final RecordReaderIterator<InternalRow> rowIterator =
-                            new RecordReaderIterator<>(reader);
+                                tableRead.executeFilter().createReader(split.getSplit());
+                        final RecordReaderIterator<InternalRow> rowIterator =
+                                new RecordReaderIterator<>(reader)) {
                     while (rowIterator.hasNext()) {
                         final InternalRow row = rowIterator.next();
                         final SeaTunnelRow seaTunnelRow =

seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/bucket/PaimonBucketAssignerTest.java

@@ -58,7 +58,8 @@ public void before() throws Exception {
         Catalog catalog = CatalogFactory.createCatalog(CatalogContext.create(options));
         catalog.createDatabase(DATABASE_NAME, true);
         Identifier identifier = Identifier.create(DATABASE_NAME, TABLE_NAME);
-        if (!catalog.tableExists(identifier)) {
+        List<String> tables = catalog.listTables(DATABASE_NAME);
+        if (!tables.contains(identifier.getTableName())) {
             Schema.Builder schemaBuilder = Schema.newBuilder();
             schemaBuilder.column("id", DataTypes.INT(), "primary Key");
             schemaBuilder.column("name", DataTypes.STRING(), "name");

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonIT.java

@@ -19,6 +19,7 @@
 
 import org.apache.seatunnel.e2e.common.TestSuiteBase;
 import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
 import org.apache.seatunnel.e2e.common.container.TestContainer;
 import org.apache.seatunnel.e2e.common.container.TestContainerId;
 import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
@@ -34,8 +35,8 @@
 import java.nio.file.Path;
 
 @DisabledOnContainer(
-        value = TestContainerId.FLINK_1_13,
-        disabledReason = "Paimon does not support flink 1.13")
+        value = {TestContainerId.FLINK_1_13, TestContainerId.SPARK_2_4},
+        disabledReason = "Paimon does not support flink 1.13 and spark 2.x")
 public class PaimonIT extends TestSuiteBase {
 
     @TestContainerExtension
@@ -49,6 +50,11 @@ public class PaimonIT extends TestSuiteBase {
             };
 
     @TestTemplate
+    @DisabledOnContainer(
+            type = EngineType.FLINK,
+            value = {},
+            disabledReason =
+                    "this case use Local file system, e2e's flink is a distributed environment.")
     public void testWriteAndReadPaimon(TestContainer container)
             throws IOException, InterruptedException {
         Container.ExecResult textWriteResult = container.executeJob("/fake_to_paimon.conf");

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonRecordWithFullType.java

@@ -36,7 +36,7 @@ public class PaimonRecordWithFullType {
     public int[] c_array;
     public BinaryString c_string;
     public boolean c_boolean;
-    public short c_tinyint;
+    public byte c_tinyint;
     public short c_smallint;
     public int c_int;
     public long c_bigint;