fix: Fix paimon writing issue

1. Manually specify update conditions, which will cause update/delete not working;
2. Using the date type as the primary key, it will cause the update/delete not working

(cherry picked from commit b9d81be)
(cherry picked from commit e1a37cb)

Author: ply0011

connectors/paimon-connector/src/main/java/io/tapdata/connector/paimon/service/PaimonService.java

@@ -1,5 +1,7 @@
 package io.tapdata.connector.paimon.service;
 
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
 import io.tapdata.connector.paimon.config.PaimonConfig;
 import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
 import io.tapdata.entity.event.dml.TapInsertRecordEvent;
@@ -474,6 +476,7 @@ public boolean createTable(TapTable tapTable, Log log) throws Exception {
 
 		// Build schema
 		Schema.Builder schemaBuilder = Schema.newBuilder();
+		Map<String, Object> schemaBuilderVariableMap = new HashMap<>();
 
 		// Add fields
 		Map<String, TapField> fields = tapTable.getNameFieldMap();
@@ -497,19 +500,23 @@ public boolean createTable(TapTable tapTable, Log log) throws Exception {
 			// Dynamic bucket mode: set bucket to -1
 			// This mode provides better flexibility
 			schemaBuilder.option("bucket", "-1");
+			schemaBuilderVariableMap.put("bucket", -1);
 		} else {
 			// Fixed bucket mode: set specific bucket count
 			Integer bucketCount = config.getBucketCount();
 			if (bucketCount == null || bucketCount <= 0) {
 				bucketCount = 4; // Default to 4 buckets if not configured
 			}
 			schemaBuilder.option("bucket", String.valueOf(bucketCount));
+			schemaBuilderVariableMap.put("bucket", String.valueOf(bucketCount));
 		}
 		if (EmptyKit.isNotBlank(config.getFileFormat())) {
 			schemaBuilder.option("file.format", config.getFileFormat());
-		}
+			schemaBuilderVariableMap.put("file.format", config.getFileFormat());
+        }
 		if (EmptyKit.isNotBlank(config.getCompression())) {
 			schemaBuilder.option("compression", config.getCompression());
+			schemaBuilderVariableMap.put("compression", config.getCompression());
 		}
 
 		// ===== Performance Optimization Options =====
@@ -518,55 +525,74 @@ public boolean createTable(TapTable tapTable, Log log) throws Exception {
 		// Larger buffer = better performance but more memory usage
 		if (config.getWriteBufferSize() != null && config.getWriteBufferSize() > 0) {
 			schemaBuilder.option("write-buffer-size", config.getWriteBufferSize() + "mb");
+			schemaBuilderVariableMap.put("write-buffer-size", config.getWriteBufferSize() + "mb");
 		}
 
 		// 2. Target file size - Paimon will try to create files of this size
 		// Larger files = fewer files but slower compaction
 		if (config.getTargetFileSize() != null && config.getTargetFileSize() > 0) {
 			schemaBuilder.option("target-file-size", config.getTargetFileSize() + "mb");
+			schemaBuilderVariableMap.put("target-file-size", config.getTargetFileSize() + "mb");
 		}
 
 		// 3. Compaction settings
 		if (config.getEnableAutoCompaction() != null) {
 			if (config.getEnableAutoCompaction()) {
 				// Enable full compaction for better query performance
-				schemaBuilder.option("compaction.optimization-interval",
-					config.getCompactionIntervalMinutes() + "min");
+				schemaBuilder.option("compaction.async.enabled", "true");
+				schemaBuilder.option("compaction.optimization-interval", config.getCompactionIntervalMinutes() + "min");
+				schemaBuilderVariableMap.put("compaction.async.enabled", "true");
+				schemaBuilderVariableMap.put("compaction.optimization-interval", config.getCompactionIntervalMinutes() + "min");
 
 				// Set compaction strategy
-				schemaBuilder.option("changelog-producer", "full-compaction");
+				schemaBuilder.option("changelog-producer", "input");
+				schemaBuilderVariableMap.put("changelog-producer", "input");
 
 				// Compact small files more aggressively
 				schemaBuilder.option("num-sorted-run.compaction-trigger", "3");
+				schemaBuilderVariableMap.put("num-sorted-run.compaction-trigger", "3");
 				schemaBuilder.option("num-sorted-run.stop-trigger", "5");
+				schemaBuilderVariableMap.put("num-sorted-run.stop-trigger", "5");
 			} else {
 				// Disable auto compaction
 				schemaBuilder.option("compaction.optimization-interval", "0");
+				schemaBuilderVariableMap.put("compaction.optimization-interval", "0");
 			}
 		}
 
 		// 4. Snapshot settings for better performance
 		// Keep more snapshots in memory for faster access
-		schemaBuilder.option("snapshot.num-retained.min", "10");
-		schemaBuilder.option("snapshot.num-retained.max", "100");
-		schemaBuilder.option("snapshot.time-retained", "1h");
+		schemaBuilder.option("snapshot.num-retained.min", "5");
+		schemaBuilder.option("snapshot.num-retained.max", "50");
+		schemaBuilder.option("snapshot.time-retained", "30min");
+		schemaBuilderVariableMap.put("snapshot.num-retained.min", "5");
+		schemaBuilderVariableMap.put("snapshot.num-retained.max", "50");
+		schemaBuilderVariableMap.put("snapshot.time-retained", "30min");
 
 		// 5. Commit settings
 		// Force compact on commit for better read performance
 		schemaBuilder.option("commit.force-compact", "false"); // Don't force compact on every commit
+		schemaBuilderVariableMap.put("commit.force-compact", "false");
 
 		// 6. Scan settings for better read performance
 		schemaBuilder.option("scan.plan-sort-partition", "true");
+		schemaBuilderVariableMap.put("scan.plan-sort-partition", "true");
 
 		// 7. Changelog settings for CDC scenarios
 		schemaBuilder.option("changelog-producer.lookup-wait", "false"); // Don't wait for lookup
+		schemaBuilderVariableMap.put("changelog-producer.lookup-wait", "false");
 
 		// 8. Memory settings
 		schemaBuilder.option("sink.parallelism", String.valueOf(config.getWriteThreads()));
+		schemaBuilderVariableMap.put("sink.parallelism", String.valueOf(config.getWriteThreads()));
 
 		// Create table
 		catalog.createTable(identifier, schemaBuilder.build(), false);
 
+		// log schema builder variables
+		Gson gson = new GsonBuilder().setPrettyPrinting().create();
+		log.info("Created table {} with schema: {}", identifier.getFullName(), gson.toJson(schemaBuilder.build()));
+
 		return true;
 	}
 
@@ -1071,8 +1097,12 @@ private void handleStreamInsert(TapInsertRecordEvent event, StreamTableWrite wri
 		String database = config.getDatabase();
 		Identifier identifier = Identifier.create(database, table.getName());
 		GenericRow row = convertToGenericRow(after, table, identifier);
-		int bucket = selectBucketForDynamic(row, table);
-		writer.write(row, bucket);
+		if (config.getBucketMode().equals("fixed")) {
+			writer.write(row);
+		} else {
+			int bucket = selectBucketForDynamic(row, table);
+			writer.write(row, bucket);
+		}
 	}
 
 	/**
@@ -1088,8 +1118,12 @@ private void handleStreamUpdate(TapUpdateRecordEvent event, StreamTableWrite wri
 		String database = config.getDatabase();
 		Identifier identifier = Identifier.create(database, table.getName());
 		GenericRow row = convertToGenericRow(after, table, identifier);
-		int bucket = selectBucketForDynamic(row, table);
-		writer.write(row, bucket);
+		if (config.getBucketMode().equals("fixed")) {
+			writer.write(row);
+		} else {
+			int bucket = selectBucketForDynamic(row, table);
+			writer.write(row, bucket);
+		}
 	}
 
 	/**
@@ -1107,8 +1141,12 @@ private void handleStreamDelete(TapDeleteRecordEvent event, StreamTableWrite wri
 		GenericRow row = convertToGenericRow(before, table, identifier);
 		// Set row kind to DELETE
 		row.setRowKind(org.apache.paimon.types.RowKind.DELETE);
-		int bucket = selectBucketForDynamic(row, table);
-		writer.write(row, bucket);
+		if (config.getBucketMode().equals("fixed")) {
+			writer.write(row);
+		} else {
+			int bucket = selectBucketForDynamic(row, table);
+			writer.write(row, bucket);
+		}
 	}
 
 	/**