[FLINK-36524][pipeline-connector][paimon] Bump Paimon version to 0.9.0

lvyanquan · web-flow · commit d3c049d8a7a2 · 2025-01-09T12:01:52.000+08:00
This closes #3644
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml
@@ -29,7 +29,7 @@ limitations under the License.
     <artifactId>flink-cdc-pipeline-connector-paimon</artifactId>
 
     <properties>
-        <paimon.version>0.8.2</paimon.version>
+        <paimon.version>0.9.0</paimon.version>
         <hadoop.version>2.8.5</hadoop.version>
         <hive.version>2.3.9</hive.version>
         <mockito.version>3.4.6</mockito.version>
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactory.java
@@ -69,6 +69,8 @@ public DataSink createDataSink(Context context) {
                     }
                 });
         Options options = Options.fromMap(catalogOptions);
+        // Avoid using previous table schema.
+        options.setString("cache-enabled", "false");
         try (Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options)) {
             Preconditions.checkNotNull(
                     catalog.listDatabases(), "catalog option of Paimon is invalid.");
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java
@@ -45,8 +45,8 @@ public PaimonCommitter(Options catalogOptions, String commitUser) {
         storeMultiCommitter =
                 new StoreMultiCommitter(
                         () -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions),
-                        commitUser,
-                        null);
+                        org.apache.paimon.flink.sink.Committer.createContext(
+                                commitUser, null, true, false, null));
     }
 
     @Override
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/StoreSinkWriteImpl.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/StoreSinkWriteImpl.java
@@ -142,6 +142,9 @@ public void withCompactExecutor(ExecutorService compactExecutor) {
         write.withCompactExecutor(compactExecutor);
     }
 
+    @Override
+    public void withInsertOnly(boolean b) {}
+
     @Override
     public SinkRecord write(InternalRow internalRow) throws Exception {
         return write.writeAndReturn(internalRow);
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
@@ -151,26 +151,26 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
                             dataChangeEvent,
                             schemaMaps.get(dataChangeEvent.tableId()).getFieldGetters());
             switch (tuple4.f0) {
-                case DYNAMIC:
+                case HASH_DYNAMIC:
                     {
                         bucket =
                                 tuple4.f2.assign(
                                         tuple4.f3.partition(genericRow),
                                         tuple4.f3.trimmedPrimaryKey(genericRow).hashCode());
                         break;
                     }
-                case FIXED:
+                case HASH_FIXED:
                     {
                         tuple4.f1.setRecord(genericRow);
                         bucket = tuple4.f1.bucket();
                         break;
                     }
-                case UNAWARE:
+                case BUCKET_UNAWARE:
                     {
                         bucket = 0;
                         break;
                     }
-                case GLOBAL_DYNAMIC:
+                case CROSS_PARTITION:
                 default:
                     {
                         throw new RuntimeException("Unsupported bucket mode: " + tuple4.f0);
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java
@@ -87,6 +87,7 @@ private void initialize(String metastore)
         }
         catalogOptions.setString("metastore", metastore);
         catalogOptions.setString("warehouse", warehouse);
+        catalogOptions.setString("cache-enabled", "false");
         this.catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
         this.catalog.dropDatabase(TEST_DATABASE, true, true);
     }
@@ -206,6 +207,30 @@ public void testApplySchemaChange(String metastore)
                 catalog.getTable(Identifier.fromString("test.table_with_partition"));
         Assertions.assertEquals(tableSchema, tableWithPartition.rowType());
         Assertions.assertEquals(Arrays.asList("col1", "dt"), tableWithPartition.primaryKeys());
+        // Create table with upper case.
+        catalogOptions.setString("allow-upper-case", "true");
+        metadataApplier = new PaimonMetadataApplier(catalogOptions);
+        createTableEvent =
+                new CreateTableEvent(
+                        TableId.parse("test.table_with_upper_case"),
+                        org.apache.flink.cdc.common.schema.Schema.newBuilder()
+                                .physicalColumn(
+                                        "COL1",
+                                        org.apache.flink.cdc.common.types.DataTypes.STRING()
+                                                .notNull())
+                                .physicalColumn(
+                                        "col2", org.apache.flink.cdc.common.types.DataTypes.INT())
+                                .primaryKey("COL1")
+                                .build());
+        metadataApplier.applySchemaChange(createTableEvent);
+        tableSchema =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(0, "COL1", DataTypes.STRING().notNull()),
+                                new DataField(1, "col2", DataTypes.INT())));
+        Assertions.assertEquals(
+                tableSchema,
+                catalog.getTable(Identifier.fromString("test.table_with_upper_case")).rowType());
     }
 
     @ParameterizedTest
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
@@ -127,7 +127,7 @@ private void initialize(String metastore)
                                     + "'metastore'='hive', "
                                     + "'hadoop-conf-dir'='%s', "
                                     + "'hive-conf-dir'='%s', "
-                                    + "'cache-enabled'='false' "
+                                    + "'cache-enabled'='false'"
                                     + ")",
                             warehouse, HADOOP_CONF_DIR, HIVE_CONF_DIR));
         } else {

Original file line number	Diff line number	Diff line change
`@@ -69,6 +69,8 @@ public DataSink createDataSink(Context context) {`
`69`	`69`	`}`
`70`	`70`	`});`
`71`	`71`	`Options options = Options.fromMap(catalogOptions);`
	`72`	`+ // Avoid using previous table schema.`
	`73`	`+ options.setString("cache-enabled", "false");`
`72`	`74`	`try (Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options)) {`
`73`	`75`	`Preconditions.checkNotNull(`
`74`	`76`	`catalog.listDatabases(), "catalog option of Paimon is invalid.");`
Original file line number	Diff line number	Diff line change
`@@ -142,6 +142,9 @@ public void withCompactExecutor(ExecutorService compactExecutor) {`
`142`	`142`	`write.withCompactExecutor(compactExecutor);`
`143`	`143`	`}`
`144`	`144`
	`145`	`+ @Override`
	`146`	`+ public void withInsertOnly(boolean b) {}`
	`147`	`+`
`145`	`148`	`@Override`
`146`	`149`	`public SinkRecord write(InternalRow internalRow) throws Exception {`
`147`	`150`	`return write.writeAndReturn(internalRow);`
Original file line number	Diff line number	Diff line change
`@@ -151,26 +151,26 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {`
`151`	`151`	`dataChangeEvent,`
`152`	`152`	`schemaMaps.get(dataChangeEvent.tableId()).getFieldGetters());`
`153`	`153`	`switch (tuple4.f0) {`
`154`		`- case DYNAMIC:`
	`154`	`+ case HASH_DYNAMIC:`
`155`	`155`	`{`
`156`	`156`	`bucket =`
`157`	`157`	`tuple4.f2.assign(`
`158`	`158`	`tuple4.f3.partition(genericRow),`
`159`	`159`	`tuple4.f3.trimmedPrimaryKey(genericRow).hashCode());`
`160`	`160`	`break;`
`161`	`161`	`}`
`162`		`- case FIXED:`
	`162`	`+ case HASH_FIXED:`
`163`	`163`	`{`
`164`	`164`	`tuple4.f1.setRecord(genericRow);`
`165`	`165`	`bucket = tuple4.f1.bucket();`
`166`	`166`	`break;`
`167`	`167`	`}`
`168`		`- case UNAWARE:`
	`168`	`+ case BUCKET_UNAWARE:`
`169`	`169`	`{`
`170`	`170`	`bucket = 0;`
`171`	`171`	`break;`
`172`	`172`	`}`
`173`		`- case GLOBAL_DYNAMIC:`
	`173`	`+ case CROSS_PARTITION:`
`174`	`174`	`default:`
`175`	`175`	`{`
`176`	`176`	`throw new RuntimeException("Unsupported bucket mode: " + tuple4.f0);`