Checkpoint 48 - Fix partitioning issue for non-partitioned tables

Author: voonhous

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/bucket/BucketAssignOperator.java

@@ -301,21 +301,22 @@ private String extractRecordKey(
      *   <li>Formats them as "field1=value1/field2=value2" (Hive-style partitioning)
      * </ul>
      *
-     * <p>If no partition keys are defined, returns "default".
+     * <p>If no partition keys are defined, returns empty string (for unpartitioned tables).
      *
      * @param event The DataChangeEvent to extract partition from
      * @param schema The table schema containing partition key definitions
      * @param fieldGetters Field getters for extracting values (not used currently, may be needed
      *     for optimization)
-     * @return The partition path string
+     * @return The partition path string (empty string for unpartitioned tables)
      */
     private String extractPartitionPath(
             DataChangeEvent event, Schema schema, List<RecordData.FieldGetter> fieldGetters) {
 
         // Check if schema has partition keys defined
         List<String> partitionKeys = schema.partitionKeys();
         if (partitionKeys == null || partitionKeys.isEmpty()) {
-            return "default";
+            // Hudi convention: unpartitioned tables use empty string, not "default"
+            return "";
         }
 
         // Get the record data to extract from (after for INSERT/UPDATE/REPLACE, before for DELETE)

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/util/RowDataUtils.java

@@ -355,17 +355,18 @@ private static String extractRecordKeyFromDataChangeEvent(
      *   <li>Formats them as "field1=value1/field2=value2" (Hive-style partitioning)
      * </ul>
      *
-     * <p>If no partition keys are defined, returns "default".
+     * <p>If no partition keys are defined, returns empty string (for unpartitioned tables).
      *
      * @param dataChangeEvent The DataChangeEvent to extract partition from
      * @param schema The table schema containing partition key definitions
-     * @return The partition path string
+     * @return The partition path string (empty string for unpartitioned tables)
      */
     private static String extractPartitionPathFromDataChangeEvent(
             DataChangeEvent dataChangeEvent, Schema schema) {
         List<String> partitionKeys = schema.partitionKeys();
         if (partitionKeys == null || partitionKeys.isEmpty()) {
-            return "default";
+            // Hudi convention: unpartitioned tables use empty string, not "default"
+            return "";
         }
 
         // Get the record data to extract from (after for INSERT/UPDATE/REPLACE, before for DELETE)