[hotfix] Refactor fluss-offsets naming in lake snapshot to align with origin design (#1772)

luoyuxia · web-flow · commit e543d1c5282c · 2025-09-29T17:14:08.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/BucketOffsetJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/BucketOffsetJsonSerde.java
@@ -29,9 +29,9 @@ public class BucketOffsetJsonSerde
 
     public static final BucketOffsetJsonSerde INSTANCE = new BucketOffsetJsonSerde();
     private static final String PARTITION_ID = "partition_id";
-    private static final String BUCKET_ID = "bucket_id";
+    private static final String BUCKET_ID = "bucket";
     private static final String PARTITION_NAME = "partition_name";
-    private static final String LOG_OFFSET = "log_offset";
+    private static final String LOG_OFFSET = "offset";
 
     @Override
     public BucketOffset deserialize(JsonNode node) {
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/BucketOffsetJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/BucketOffsetJsonSerdeTest.java
@@ -37,8 +37,8 @@ protected BucketOffset[] createObjects() {
     @Override
     protected String[] expectedJsons() {
         return new String[] {
-            "{\"partition_id\":1,\"bucket_id\":1,\"partition_name\":\"country=eu-central/year=2023/month=12\",\"log_offset\":10}",
-            "{\"bucket_id\":2,\"log_offset\":20}"
+            "{\"partition_id\":1,\"bucket\":1,\"partition_name\":\"country=eu-central/year=2023/month=12\",\"offset\":10}",
+            "{\"bucket\":2,\"offset\":20}"
         };
     }
 }
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringITCase.java
@@ -188,7 +188,7 @@ void testTiering() throws Exception {
                         {
                             put(
                                     FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
-                                    "[{\"bucket_id\":0,\"log_offset\":3}]");
+                                    "[{\"bucket\":0,\"offset\":3}]");
                         }
                     };
             checkSnapshotPropertyInIceberg(t1, properties);
@@ -402,8 +402,8 @@ private void testPartitionedTableTiering() throws Exception {
                         put(
                                 FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
                                 "["
-                                        + "{\"partition_id\":0,\"bucket_id\":0,\"partition_name\":\"date=2025\",\"log_offset\":3},"
-                                        + "{\"partition_id\":1,\"bucket_id\":0,\"partition_name\":\"date=2026\",\"log_offset\":3}"
+                                        + "{\"partition_id\":0,\"bucket\":0,\"partition_name\":\"date=2025\",\"offset\":3},"
+                                        + "{\"partition_id\":1,\"bucket\":0,\"partition_name\":\"date=2026\",\"offset\":3}"
                                         + "]");
                     }
                 };
diff --git a/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringITCase.java b/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringITCase.java
@@ -102,7 +102,7 @@ void testTiering() throws Exception {
                     {
                         put(
                                 FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
-                                "[{\"bucket_id\":0,\"log_offset\":30}]");
+                                "[{\"bucket\":0,\"offset\":30}]");
                         put("commit-user", FLUSS_LAKE_TIERING_COMMIT_USER);
                     }
                 };
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java
@@ -104,7 +104,7 @@ void testTiering() throws Exception {
                         {
                             put(
                                     FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
-                                    "[{\"bucket_id\":0,\"log_offset\":3}]");
+                                    "[{\"bucket\":0,\"offset\":3}]");
                         }
                     };
             checkSnapshotPropertyInPaimon(t1, properties);
@@ -219,7 +219,7 @@ void testTieringForAlterTable() throws Exception {
                         {
                             put(
                                     FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
-                                    "[{\"bucket_id\":0,\"log_offset\":3}]");
+                                    "[{\"bucket\":0,\"offset\":3}]");
                         }
                     };
             checkSnapshotPropertyInPaimon(t1, properties);
@@ -315,7 +315,7 @@ void testTieringForAlterTable() throws Exception {
 
     private String getPartitionOffsetStr(Map<Long, String> partitionNameByIds) {
         String raw =
-                "{\"partition_id\":%s,\"bucket_id\":0,\"partition_name\":\"date=%s\",\"log_offset\":3}";
+                "{\"partition_id\":%s,\"bucket\":0,\"partition_name\":\"date=%s\",\"offset\":3}";
         List<Long> partitionIds = new ArrayList<>(partitionNameByIds.keySet());
         Collections.sort(partitionIds);
         List<String> partitionOffsetStrs = new ArrayList<>();
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java
@@ -389,8 +389,8 @@ void testThreePartitionTiering() throws Exception {
         String offsetProperty = getSnapshotLogOffsetProperty(tablePath, snapshot);
         assertThat(offsetProperty)
                 .isEqualTo(
-                        "[{\"partition_id\":1,\"bucket_id\":0,\"partition_name\":\"region=us-east/year=2024/month=01\",\"log_offset\":2},"
-                                + "{\"partition_id\":2,\"bucket_id\":0,\"partition_name\":\"region=eu-central/year=2023/month=12\",\"log_offset\":2}]");
+                        "[{\"partition_id\":1,\"bucket\":0,\"partition_name\":\"region=us-east/year=2024/month=01\",\"offset\":2},"
+                                + "{\"partition_id\":2,\"bucket\":0,\"partition_name\":\"region=eu-central/year=2023/month=12\",\"offset\":2}]");
 
         // Verify data for each partition
         for (String partition : partitionIdAndName.values()) {
diff --git a/website/docs/streaming-lakehouse/integrate-data-lakes/paimon.md b/website/docs/streaming-lakehouse/integrate-data-lakes/paimon.md
@@ -174,4 +174,53 @@ The following table shows the mapping between [Fluss data types](table-design/da
 | TIMESTAMP                     | TIMESTAMP                     |
 | TIMESTAMP WITH LOCAL TIMEZONE | TIMESTAMP WITH LOCAL TIMEZONE |
 | BINARY                        | BINARY                        |
-| BYTES                         | BYTES                         |
+| BYTES                         | BYTES                         |
+
+## 📊 Snapshot Metadata
+
+Fluss adds specific metadata to Paimon snapshots for traceability:
+
+- **commit-user**: Set to `__fluss_lake_tiering` to identify Fluss-generated snapshots
+- **fluss-offsets**: JSON string containing the Fluss bucket offset mapping to track the tiering progress
+
+#### Non-Partitioned Tables
+
+For non-partitioned tables, the metadata structure of `fluss-offsets` is:
+
+```json
+[
+  {"bucket": 0, "offset": 1234},
+  {"bucket": 1, "offset": 5678},
+  {"bucket": 2, "offset": 9012}
+]
+```
+
+#### Partitioned Tables
+
+For partitioned tables, the metadata structure includes partition information:
+
+```json
+[
+  {
+    "partition_name": "date=2025",
+    "partition_id": 0,
+    "bucket": 0,
+    "offset": 3
+  },
+  {
+    "partition_name": "date=2025",
+    "partition_id": 1,
+    "bucket": 0,
+    "offset": 3
+  }
+]
+```
+
+#### Metadata Fields Explanation
+
+| Field            | Description                                  | Example                      |
+|------------------|----------------------------------------------|------------------------------|
+| `partition_id`   | Unique identifier in Fluss for the partition | `0`, `1`                     |
+| `bucket`         | Bucket identifier within the partition       | `0`, `1`, `2`                |
+| `partition_name` | Human-readable partition name                | `"date=2025"`, `"date=2026"` |
+| `offset`         | Offset within the partition's log            | `3`, `1000`                  |

Original file line number	Diff line number	Diff line change
`@@ -37,8 +37,8 @@ protected BucketOffset[] createObjects() {`
`37`	`37`	`@Override`
`38`	`38`	`protected String[] expectedJsons() {`
`39`	`39`	`return new String[] {`
`40`		`- "{\"partition_id\":1,\"bucket_id\":1,\"partition_name\":\"country=eu-central/year=2023/month=12\",\"log_offset\":10}",`
`41`		`- "{\"bucket_id\":2,\"log_offset\":20}"`
	`40`	`+ "{\"partition_id\":1,\"bucket\":1,\"partition_name\":\"country=eu-central/year=2023/month=12\",\"offset\":10}",`
	`41`	`+ "{\"bucket\":2,\"offset\":20}"`
`42`	`42`	`};`
`43`	`43`	`}`
`44`	`44`	`}`
Original file line number	Diff line number	Diff line change
`@@ -188,7 +188,7 @@ void testTiering() throws Exception {`
`188`	`188`	`{`
`189`	`189`	`put(`
`190`	`190`	`FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,`
`191`		`- "[{\"bucket_id\":0,\"log_offset\":3}]");`
	`191`	`+ "[{\"bucket\":0,\"offset\":3}]");`
`192`	`192`	`}`
`193`	`193`	`};`
`194`	`194`	`checkSnapshotPropertyInIceberg(t1, properties);`
`@@ -402,8 +402,8 @@ private void testPartitionedTableTiering() throws Exception {`
`402`	`402`	`put(`
`403`	`403`	`FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,`
`404`	`404`	`"["`
`405`		`- + "{\"partition_id\":0,\"bucket_id\":0,\"partition_name\":\"date=2025\",\"log_offset\":3},"`
`406`		`- + "{\"partition_id\":1,\"bucket_id\":0,\"partition_name\":\"date=2026\",\"log_offset\":3}"`
	`405`	`+ + "{\"partition_id\":0,\"bucket\":0,\"partition_name\":\"date=2025\",\"offset\":3},"`
	`406`	`+ + "{\"partition_id\":1,\"bucket\":0,\"partition_name\":\"date=2026\",\"offset\":3}"`
`407`	`407`	`+ "]");`
`408`	`408`	`}`
`409`	`409`	`};`
Original file line number	Diff line number	Diff line change
`@@ -102,7 +102,7 @@ void testTiering() throws Exception {`
`102`	`102`	`{`
`103`	`103`	`put(`
`104`	`104`	`FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,`
`105`		`- "[{\"bucket_id\":0,\"log_offset\":30}]");`
	`105`	`+ "[{\"bucket\":0,\"offset\":30}]");`
`106`	`106`	`put("commit-user", FLUSS_LAKE_TIERING_COMMIT_USER);`
`107`	`107`	`}`
`108`	`108`	`};`
Original file line number	Diff line number	Diff line change
`@@ -104,7 +104,7 @@ void testTiering() throws Exception {`
`104`	`104`	`{`
`105`	`105`	`put(`
`106`	`106`	`FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,`
`107`		`- "[{\"bucket_id\":0,\"log_offset\":3}]");`
	`107`	`+ "[{\"bucket\":0,\"offset\":3}]");`
`108`	`108`	`}`
`109`	`109`	`};`
`110`	`110`	`checkSnapshotPropertyInPaimon(t1, properties);`
`@@ -219,7 +219,7 @@ void testTieringForAlterTable() throws Exception {`
`219`	`219`	`{`
`220`	`220`	`put(`
`221`	`221`	`FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,`
`222`		`- "[{\"bucket_id\":0,\"log_offset\":3}]");`
	`222`	`+ "[{\"bucket\":0,\"offset\":3}]");`
`223`	`223`	`}`
`224`	`224`	`};`
`225`	`225`	`checkSnapshotPropertyInPaimon(t1, properties);`
`@@ -315,7 +315,7 @@ void testTieringForAlterTable() throws Exception {`
`315`	`315`
`316`	`316`	`private String getPartitionOffsetStr(Map<Long, String> partitionNameByIds) {`
`317`	`317`	`String raw =`
`318`		`- "{\"partition_id\":%s,\"bucket_id\":0,\"partition_name\":\"date=%s\",\"log_offset\":3}";`
	`318`	`+ "{\"partition_id\":%s,\"bucket\":0,\"partition_name\":\"date=%s\",\"offset\":3}";`
`319`	`319`	`List<Long> partitionIds = new ArrayList<>(partitionNameByIds.keySet());`
`320`	`320`	`Collections.sort(partitionIds);`
`321`	`321`	`List<String> partitionOffsetStrs = new ArrayList<>();`