upgrade lance version to utilize txn property

Author: xx789633

fluss-lake/fluss-lake-lance/pom.xml

@@ -33,7 +33,7 @@
     <packaging>jar</packaging>
 
     <properties>
-        <lance.version>0.32.1</lance.version>
+        <lance.version>0.33.0</lance.version>
     </properties>
 
     <dependencies>

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

@@ -104,7 +104,7 @@ public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSna
                 new CommittedLakeSnapshot(latestLakeSnapshotIdOfLake.get());
         String flussOffsetProperties =
                 LanceDatasetAdapter.getTransactionProperties(
-                                config, Math.toIntExact(latestLakeSnapshotIdOfLake.get()))
+                                config, Math.toIntExact(latestLakeSnapshotIdOfLake.get() + 1))
                         .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY);
         for (JsonNode node : OBJECT_MAPPER.readTree(flussOffsetProperties)) {
             BucketOffset bucketOffset = BucketOffsetJsonSerde.INSTANCE.deserialize(node);

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

@@ -36,6 +36,7 @@
 import org.apache.arrow.vector.ipc.ArrowReader;
 import org.apache.arrow.vector.types.pojo.Schema;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -82,23 +83,29 @@ public static long commitAppend(
             LanceConfig config, List<FragmentMetadata> fragments, Map<String, String> properties) {
         String uri = config.getDatasetUri();
         ReadOptions options = LanceConfig.genReadOptionFromConfig(config);
-        Dataset dataset = Dataset.open(allocator, uri, options);
-        Transaction transaction =
-                dataset.newTransactionBuilder()
-                        .operation(Append.builder().fragments(fragments).build())
-                        .transactionProperties(properties)
-                        .build();
-        try (Dataset appendedDataset = transaction.commit()) {
-            return appendedDataset.version() - 1;
+        try (Dataset dataset = Dataset.open(allocator, uri, options)) {
+            Transaction transaction =
+                    dataset.newTransactionBuilder()
+                            .operation(Append.builder().fragments(fragments).build())
+                            .transactionProperties(properties)
+                            .build();
+            try (Dataset appendedDataset = transaction.commit()) {
+                return appendedDataset.version() - 1;
+            }
         }
     }
 
     public static Map<String, String> getTransactionProperties(LanceConfig config, int version) {
         ReadOptions.Builder builder = new ReadOptions.Builder();
         builder.setVersion(version);
         builder.setStorageOptions(LanceConfig.genStorageOptions(config));
-        Dataset dataset = Dataset.open(allocator, config.getDatasetUri(), builder.build());
-        dataset.readTransaction();
+        try (Dataset dataset = Dataset.open(allocator, config.getDatasetUri(), builder.build())) {
+            Transaction transaction = dataset.readTransaction().orElse(null);
+            if (transaction != null) {
+                return transaction.transactionProperties();
+            }
+            return Collections.emptyMap();
+        }
     }
 
     public static Optional<Long> getVersion(LanceConfig config) {

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

@@ -93,7 +93,7 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
                         customProperties,
                         tablePath.getDatabaseName(),
                         tablePath.getTableName());
-        Schema schema = createTable(tablePath, isPartitioned, null, config);
+        Schema schema = createTable(config);
 
         List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
         SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
@@ -118,7 +118,7 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
                             }
                         }
                         : Collections.singletonMap(null, null);
-        List<String> partitionKeys = new ArrayList<>(partitionIdAndName.values());
+        List<String> partitionKeys = isPartitioned ? List.of("c3") : null;
         Map<TableBucket, Long> tableBucketOffsets = new HashMap<>();
         // first, write data
         for (int bucket = 0; bucket < bucketNum; bucket++) {
@@ -196,8 +196,8 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
             Map<Tuple2<Long, Integer>, Long> offsets = committedLakeSnapshot.getLogEndOffsets();
             for (int bucket = 0; bucket < 3; bucket++) {
                 for (Long partitionId : partitionIdAndName.keySet()) {
-                    // we only write 10 records, so expected log offset should be 9
-                    assertThat(offsets.get(Tuple2.of(partitionId, bucket))).isEqualTo(9);
+                    // we only write 10 records, so expected log offset should be 10
+                    assertThat(offsets.get(Tuple2.of(partitionId, bucket))).isEqualTo(10);
                 }
             }
             assertThat(committedLakeSnapshot.getLakeSnapshotId()).isOne();
@@ -303,24 +303,18 @@ private Tuple2<List<LogRecord>, List<LogRecord>> genLogTableRecords(
         return Tuple2.of(logRecords, logRecords);
     }
 
-    private Schema createTable(
-            TablePath tablePath,
-            boolean isPartitioned,
-            @Nullable Integer numBuckets,
-            LanceConfig config)
-            throws Exception {
+    private Schema createTable(LanceConfig config) throws Exception {
         List<Schema.Column> columns = new ArrayList<>();
         columns.add(new Schema.Column("c1", DataTypes.INT()));
         columns.add(new Schema.Column("c2", DataTypes.STRING()));
         columns.add(new Schema.Column("c3", DataTypes.STRING()));
         Schema.Builder schemaBuilder = Schema.newBuilder().fromColumns(columns);
         Schema schema = schemaBuilder.build();
-        doCreateLanceTable(tablePath, schema, config);
+        doCreateLanceTable(schema, config);
         return schema;
     }
 
-    private void doCreateLanceTable(TablePath tablePath, Schema schema, LanceConfig config)
-            throws Exception {
+    private void doCreateLanceTable(Schema schema, LanceConfig config) throws Exception {
         WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
         LanceDatasetAdapter.createDataset(
                 config.getDatasetUri(), LanceArrowUtils.toArrowSchema(schema.getRowType()), params);