[FLINK-36609][pipeline-connector][paimon] Add partition columns to primary keys

This closes  #3641.

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java

@@ -164,14 +164,22 @@ private void applyCreateTable(CreateTableEvent event) throws SchemaEvolveExcepti
                                             LogicalTypeConversion.toDataType(
                                                     DataTypeUtils.toFlinkDataType(column.getType())
                                                             .getLogicalType())));
-            builder.primaryKey(schema.primaryKeys().toArray(new String[0]));
+            List<String> partitionKeys = new ArrayList<>();
+            List<String> primaryKeys = schema.primaryKeys();
             if (partitionMaps.containsKey(event.tableId())) {
-                builder.partitionKeys(partitionMaps.get(event.tableId()));
+                partitionKeys.addAll(partitionMaps.get(event.tableId()));
             } else if (schema.partitionKeys() != null && !schema.partitionKeys().isEmpty()) {
-                builder.partitionKeys(schema.partitionKeys());
+                partitionKeys.addAll(schema.partitionKeys());
             }
-            builder.options(tableOptions);
-            builder.options(schema.options());
+            for (String partitionColumn : partitionKeys) {
+                if (!primaryKeys.contains(partitionColumn)) {
+                    primaryKeys.add(partitionColumn);
+                }
+            }
+            builder.partitionKeys(partitionKeys)
+                    .primaryKey(primaryKeys)
+                    .options(tableOptions)
+                    .options(schema.options());
             catalog.createTable(
                     new Identifier(event.tableId().getSchemaName(), event.tableId().getTableName()),
                     builder.build(),

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java

@@ -177,6 +177,35 @@ public void testApplySchemaChange(String metastore)
                                 new DataField(2, "newcol3", DataTypes.STRING())));
         Assertions.assertEquals(
                 tableSchema, catalog.getTable(Identifier.fromString("test.table1")).rowType());
+
+        // Create table with partition column.
+        createTableEvent =
+                new CreateTableEvent(
+                        TableId.parse("test.table_with_partition"),
+                        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())
+                                .physicalColumn(
+                                        "dt",
+                                        org.apache.flink.cdc.common.types.DataTypes.INT().notNull())
+                                .primaryKey("col1")
+                                .partitionKey("dt")
+                                .build());
+        metadataApplier.applySchemaChange(createTableEvent);
+        tableSchema =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(0, "col1", DataTypes.STRING().notNull()),
+                                new DataField(1, "col2", DataTypes.INT()),
+                                new DataField(2, "dt", DataTypes.INT().notNull())));
+        Table tableWithPartition =
+                catalog.getTable(Identifier.fromString("test.table_with_partition"));
+        Assertions.assertEquals(tableSchema, tableWithPartition.rowType());
+        Assertions.assertEquals(Arrays.asList("col1", "dt"), tableWithPartition.primaryKeys());
     }
 
     @ParameterizedTest
@@ -217,10 +246,10 @@ public void testCreateTableWithOptions(String metastore)
                         Arrays.asList(
                                 new DataField(0, "col1", DataTypes.STRING().notNull()),
                                 new DataField(1, "col2", DataTypes.STRING()),
-                                new DataField(2, "col3", DataTypes.STRING()),
-                                new DataField(3, "col4", DataTypes.STRING())));
+                                new DataField(2, "col3", DataTypes.STRING().notNull()),
+                                new DataField(3, "col4", DataTypes.STRING().notNull())));
         Assertions.assertEquals(tableSchema, table.rowType());
-        Assertions.assertEquals(Collections.singletonList("col1"), table.primaryKeys());
+        Assertions.assertEquals(Arrays.asList("col1", "col3", "col4"), table.primaryKeys());
         Assertions.assertEquals(Arrays.asList("col3", "col4"), table.partitionKeys());
         Assertions.assertEquals("-1", table.options().get("bucket"));
     }