commit snapshot version from 2

Author: xx789633

fluss-lake/fluss-lake-lance/src/main/java/com/alibaba/fluss/lake/lance/tiering/LanceLakeCommitter.java

@@ -34,6 +34,7 @@
 import com.lancedb.lance.Transaction;
 import com.lancedb.lance.Version;
 import org.apache.arrow.memory.RootAllocator;
+import org.apache.commons.lang3.tuple.ImmutablePair;
 
 import javax.annotation.Nullable;
 
@@ -93,25 +94,26 @@ public void abort(LanceCommittable committable) throws IOException {
     @Override
     public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSnapshotIdOfFluss)
             throws IOException {
-        Transaction latestLakeSnapshotIdOfLake =
+        ImmutablePair<Version, Transaction> latestLakeSnapshotIdOfLake =
                 getCommittedLatestSnapshotOfLake(FLUSS_LAKE_TIERING_COMMIT_USER);
 
-        if (latestLakeSnapshotIdOfLake == null) {
+        if (latestLakeSnapshotIdOfLake.equals(ImmutablePair.nullPair())) {
             return null;
         }
 
         // we get the latest snapshot committed by fluss,
         // but the latest snapshot is not greater than latestLakeSnapshotIdOfFluss, no any missing
         // snapshot, return directly
         if (latestLakeSnapshotIdOfFluss != null
-                && latestLakeSnapshotIdOfLake.readVersion() <= latestLakeSnapshotIdOfFluss) {
+                && latestLakeSnapshotIdOfLake.getLeft().getId() <= latestLakeSnapshotIdOfFluss) {
             return null;
         }
 
         CommittedLakeSnapshot committedLakeSnapshot =
-                new CommittedLakeSnapshot(latestLakeSnapshotIdOfLake.readVersion());
+                new CommittedLakeSnapshot(latestLakeSnapshotIdOfLake.getLeft().getId());
         String flussOffsetProperties =
                 latestLakeSnapshotIdOfLake
+                        .getRight()
                         .transactionProperties()
                         .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY);
         for (JsonNode node : OBJECT_MAPPER.readTree(flussOffsetProperties)) {
@@ -131,8 +133,8 @@ public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSna
     }
 
     @Nullable
-    private Transaction getCommittedLatestSnapshotOfLake(String commitUser) {
-        Transaction latestFlussSnapshot = null;
+    private ImmutablePair<Version, Transaction> getCommittedLatestSnapshotOfLake(
+            String commitUser) {
         ReadOptions.Builder builder = new ReadOptions.Builder();
         builder.setStorageOptions(LanceConfig.genStorageOptions(config));
         try (Dataset dataset = Dataset.open(allocator, config.getDatasetUri(), builder.build())) {
@@ -146,13 +148,12 @@ private Transaction getCommittedLatestSnapshotOfLake(String commitUser) {
                     if (transaction != null
                             && commitUser.equals(
                                     transaction.transactionProperties().get(committerName))) {
-                        latestFlussSnapshot = transaction;
-                        break;
+                        return ImmutablePair.of(version, transaction);
                     }
                 }
             }
         }
-        return latestFlussSnapshot;
+        return ImmutablePair.nullPair();
     }
 
     @Override

fluss-lake/fluss-lake-lance/src/main/java/com/alibaba/fluss/lake/lance/utils/LanceArrowUtils.java

@@ -224,16 +224,7 @@ protected ArrowType defaultMethod(DataType dataType) {
     }
 
     private static int getPrecision(DecimalVector decimalVector) {
-        int precision = -1;
-        try {
-            java.lang.reflect.Field precisionField =
-                    decimalVector.getClass().getDeclaredField("precision");
-            precisionField.setAccessible(true);
-            precision = (int) precisionField.get(decimalVector);
-        } catch (NoSuchFieldException | IllegalAccessException e) {
-            // should not happen, ignore
-        }
-        return precision;
+        return decimalVector.getPrecision();
     }
 
     public static ArrowFieldWriter<InternalRow> createArrowFieldWriter(

fluss-lake/fluss-lake-lance/src/main/java/com/alibaba/fluss/lake/lance/utils/LanceDatasetAdapter.java

@@ -69,8 +69,8 @@ public static long commitAppend(
                             .transactionProperties(properties)
                             .build();
             try (Dataset appendedDataset = transaction.commit()) {
-                // lance dataset version starts from 1
-                return appendedDataset.version() - 1;
+                // note: lance dataset version starts from 1
+                return appendedDataset.version();
             }
         }
     }

fluss-lake/fluss-lake-lance/src/test/java/com/alibaba/fluss/lake/lance/tiering/LanceTieringTest.java

@@ -105,7 +105,7 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
         try (LakeCommitter<LanceWriteResult, LanceCommittable> lakeCommitter =
                 createLakeCommitter(tablePath)) {
             // should no any missing snapshot
-            assertThat(lakeCommitter.getMissingLakeSnapshot(1L)).isNull();
+            assertThat(lakeCommitter.getMissingLakeSnapshot(2L)).isNull();
         }
 
         Map<Tuple2<String, Integer>, List<LogRecord>> recordsByBucket = new HashMap<>();
@@ -161,7 +161,8 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
                             lanceCommittable,
                             toBucketOffsetsProperty(
                                     tableBucketOffsets, partitionIdAndName, partitionKeys));
-            assertThat(snapshot).isEqualTo(1);
+            // lance dataset version starts from 1
+            assertThat(snapshot).isEqualTo(2);
         }
 
         try (Dataset dataset =
@@ -188,8 +189,8 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
         // then, let's verify getMissingLakeSnapshot works
         try (LakeCommitter<LanceWriteResult, LanceCommittable> lakeCommitter =
                 createLakeCommitter(tablePath)) {
-            // use snapshot id 0 as the known snapshot id
-            CommittedLakeSnapshot committedLakeSnapshot = lakeCommitter.getMissingLakeSnapshot(0L);
+            // use snapshot id 1 as the known snapshot id
+            CommittedLakeSnapshot committedLakeSnapshot = lakeCommitter.getMissingLakeSnapshot(1L);
             assertThat(committedLakeSnapshot).isNotNull();
             Map<Tuple2<Long, Integer>, Long> offsets = committedLakeSnapshot.getLogEndOffsets();
             for (int bucket = 0; bucket < 3; bucket++) {
@@ -198,16 +199,16 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
                     assertThat(offsets.get(Tuple2.of(partitionId, bucket))).isEqualTo(10);
                 }
             }
-            assertThat(committedLakeSnapshot.getLakeSnapshotId()).isOne();
+            assertThat(committedLakeSnapshot.getLakeSnapshotId()).isEqualTo(2L);
 
             // use null as the known snapshot id
             CommittedLakeSnapshot committedLakeSnapshot2 =
                     lakeCommitter.getMissingLakeSnapshot(null);
             assertThat(committedLakeSnapshot2).isEqualTo(committedLakeSnapshot);
 
-            // use snapshot id 1 as the known snapshot id
-            committedLakeSnapshot = lakeCommitter.getMissingLakeSnapshot(1L);
-            // no any missing committed offset since the latest snapshot is 1L
+            // use snapshot id 2 as the known snapshot id
+            committedLakeSnapshot = lakeCommitter.getMissingLakeSnapshot(2L);
+            // no any missing committed offset since the latest snapshot is 2L
             assertThat(committedLakeSnapshot).isNull();
         }
     }