address yuxia's comments

Author: LiebingYu

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java

@@ -84,7 +84,7 @@
  * <p>The enumerator is responsible for:
  *
  * <ul>
- *   <li>Get the all splits(lake split + snapshot split + log split) for a table of Fluss to be
+ *   <li>Get the all splits(lake split + kv snapshot split + log split) for a table of Fluss to be
  *       read.
  *   <li>Assign the splits to readers with the guarantee that the splits belong to the same bucket
  *       will be assigned to same reader.

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java

@@ -772,12 +772,17 @@ void testUnionReadPartitionsExistInPaimonButExpiredInFluss() throws Exception {
         // first of all, start tiering
         JobClient jobClient = buildTieringJob(execEnv);
 
-        String tableName = "expired_partition_pkTable";
+        String tableName = "expired_partition_pk_table";
         TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+        // create table and write data
         Map<TableBucket, Long> bucketLogEndOffset = new HashMap<>();
-        // create table & write initial data
+        Function<String, List<InternalRow>> rowGenerator =
+                (partition) ->
+                        Arrays.asList(
+                                row(3, "string", partition), row(30, "another_string", partition));
         long tableId =
-                preparePKTableFullType(tablePath, DEFAULT_BUCKET_NUM, true, bucketLogEndOffset);
+                prepareSimplePKTable(
+                        tablePath, DEFAULT_BUCKET_NUM, true, rowGenerator, bucketLogEndOffset);
 
         // wait until records has been synced
         waitUntilBucketSynced(tablePath, tableId, DEFAULT_BUCKET_NUM, true);
@@ -787,62 +792,31 @@ void testUnionReadPartitionsExistInPaimonButExpiredInFluss() throws Exception {
         assertThat(partitionNameByIds.size()).isGreaterThan(0);
 
         // Build expected rows for all partitions
+        // Simple PK table has 3 columns: c1 (INT), c2 (STRING), c3 (STRING) where c3 is partition
         List<Row> expectedAllRows = new ArrayList<>();
         for (String partition : partitionNameByIds.values()) {
-            expectedAllRows.add(
-                    Row.of(
-                            false,
-                            (byte) 1,
-                            (short) 2,
-                            3,
-                            4L,
-                            5.1f,
-                            6.0d,
-                            "string",
-                            Decimal.fromUnscaledLong(9, 5, 2),
-                            Decimal.fromBigDecimal(new java.math.BigDecimal(10), 20, 0),
-                            TimestampLtz.fromEpochMillis(1698235273182L),
-                            TimestampLtz.fromEpochMillis(1698235273182L, 5000),
-                            TimestampNtz.fromMillis(1698235273183L),
-                            TimestampNtz.fromMillis(1698235273183L, 6000),
-                            new byte[] {1, 2, 3, 4},
-                            partition));
-            expectedAllRows.add(
-                    Row.of(
-                            true,
-                            (byte) 10,
-                            (short) 20,
-                            30,
-                            40L,
-                            50.1f,
-                            60.0d,
-                            "another_string",
-                            Decimal.fromUnscaledLong(90, 5, 2),
-                            Decimal.fromBigDecimal(new java.math.BigDecimal(100), 20, 0),
-                            TimestampLtz.fromEpochMillis(1698235273200L),
-                            TimestampLtz.fromEpochMillis(1698235273200L, 5000),
-                            TimestampNtz.fromMillis(1698235273201L),
-                            TimestampNtz.fromMillis(1698235273201L, 6000),
-                            new byte[] {1, 2, 3, 4},
-                            partition));
+            expectedAllRows.add(Row.of(3, "string", partition));
+            expectedAllRows.add(Row.of(30, "another_string", partition));
         }
 
         // Select one partition to drop (expire in Fluss)
         Long partitionToDropId = partitionNameByIds.keySet().iterator().next();
         String partitionToDropName = partitionNameByIds.get(partitionToDropId);
 
         // Filter rows that belong to the partition to be dropped
+        // c3 is the partition column (index 2)
         List<Row> rowsInExpiredPartition =
                 expectedAllRows.stream()
-                        .filter(row -> partitionToDropName.equals(row.getField(15)))
+                        .filter(row -> partitionToDropName.equals(row.getField(2)))
                         .collect(Collectors.toList());
         assertThat(rowsInExpiredPartition).isNotEmpty();
 
         // Now drop the partition in Fluss (make it expired)
         // The partition data still exists in Paimon
+        // c3 is the partition column for simple PK table
         admin.dropPartition(
                         tablePath,
-                        new PartitionSpec(Collections.singletonMap("c16", partitionToDropName)),
+                        new PartitionSpec(Collections.singletonMap("c3", partitionToDropName)),
                         false)
                 .get();
 
@@ -871,12 +845,13 @@ void testUnionReadPartitionsExistInPaimonButExpiredInFluss() throws Exception {
                         expectedAllRows.stream().map(Row::toString).collect(Collectors.toList()));
 
         // Test partition filter - query only the expired partition
+        // c3 is the partition column for simple PK table
         String sqlWithPartitionFilter =
                 "select"
                         + " /*+ OPTIONS('scan.partition.discovery.interval'='100ms') */"
                         + " * FROM "
                         + tableName
-                        + " WHERE c16 = '"
+                        + " WHERE c3 = '"
                         + partitionToDropName
                         + "'";
         iterator = streamTEnv.executeSql(sqlWithPartitionFilter).collect();