address comments

Author: xx789633

fluss-lake/fluss-lake-lance/src/main/java/com/alibaba/fluss/lake/lance/LanceConfig.java

@@ -32,8 +32,6 @@ public class LanceConfig implements Serializable {
 
     private static final String block_size = "block_size";
     private static final String version = "version";
-    private static final String index_cache_size = "index_cache_size";
-    private static final String metadata_cache_size = "metadata_cache_size";
     private static final String max_row_per_file = "max_row_per_file";
     private static final String max_rows_per_group = "max_rows_per_group";
     private static final String max_bytes_per_file = "max_bytes_per_file";
@@ -83,14 +81,6 @@ public Map<String, String> getOptions() {
         return options;
     }
 
-    public String getDatabaseName() {
-        return databaseName;
-    }
-
-    public String getTableName() {
-        return tableName;
-    }
-
     public String getDatasetUri() {
         return datasetUri;
     }
@@ -104,12 +94,6 @@ public static ReadOptions genReadOptionFromConfig(LanceConfig config) {
         if (maps.containsKey(version)) {
             builder.setVersion(Integer.parseInt(maps.get(version)));
         }
-        if (maps.containsKey(index_cache_size)) {
-            builder.setIndexCacheSize(Integer.parseInt(maps.get(index_cache_size)));
-        }
-        if (maps.containsKey(metadata_cache_size)) {
-            builder.setMetadataCacheSize(Integer.parseInt(maps.get(metadata_cache_size)));
-        }
         builder.setStorageOptions(genStorageOptions(config));
         return builder.build();
     }

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

@@ -39,11 +39,11 @@ public int getVersion() {
 
     @Override
     public byte[] serialize(LanceCommittable lanceCommittable) throws IOException {
-        ByteArrayOutputStream baos = new ByteArrayOutputStream();
-        ObjectOutputStream oos = new ObjectOutputStream(baos);
-        oos.writeObject(lanceCommittable.committable());
-        oos.close();
-        return baos.toByteArray();
+        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+            oos.writeObject(lanceCommittable.committable());
+            return baos.toByteArray();
+        }
     }
 
     @Override
@@ -56,9 +56,9 @@ public LanceCommittable deserialize(int version, byte[] serialized) throws IOExc
                             + version
                             + ".");
         }
-        ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
-        ObjectInputStream ois = new ObjectInputStream(bais);
-        try {
+        try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+                ObjectInputStream ois = new ObjectInputStream(bais)) {
+            //noinspection unchecked
             return new LanceCommittable((List<FragmentMetadata>) ois.readObject());
         } catch (ClassNotFoundException e) {
             throw new IOException("Couldn't deserialize LanceCommittable", e);

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

@@ -50,6 +50,7 @@
 /** Implementation of {@link LakeCommitter} for Lance. */
 public class LanceLakeCommitter implements LakeCommitter<LanceWriteResult, LanceCommittable> {
     private final LanceConfig config;
+    private static String commitUser = "commit-user";
     private final RootAllocator allocator = new RootAllocator();
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
@@ -77,12 +78,15 @@ public LanceCommittable toCommittable(List<LanceWriteResult> lanceWriteResults)
     public long commit(LanceCommittable committable, Map<String, String> snapshotProperties)
             throws IOException {
         Map<String, String> properties = new HashMap<>(snapshotProperties);
-        properties.put("commit-user", FLUSS_LAKE_TIERING_COMMIT_USER);
+        properties.put(commitUser, FLUSS_LAKE_TIERING_COMMIT_USER);
         return LanceDatasetAdapter.commitAppend(config, committable.committable(), properties);
     }
 
     @Override
-    public void abort(LanceCommittable committable) throws IOException {}
+    public void abort(LanceCommittable committable) throws IOException {
+        // TODO lance does not have the API to proactively delete the written files yet, see
+        // https://github.com/lancedb/lance/issues/4508
+    }
 
     @SuppressWarnings("checkstyle:LocalVariableName")
     @Nullable
@@ -140,7 +144,7 @@ private Transaction getCommittedLatestSnapshotOfLake(String commitUser) {
                     Transaction transaction = datasetVersion.readTransaction().orElse(null);
                     if (transaction != null
                             && commitUser.equals(
-                                    transaction.transactionProperties().get("commit-user"))) {
+                                    transaction.transactionProperties().get(commitUser))) {
                         if (latestFlussSnapshot == null
                                 || transaction.readVersion() > latestFlussSnapshot.readVersion()) {
                             latestFlussSnapshot = transaction;
@@ -153,5 +157,7 @@ private Transaction getCommittedLatestSnapshotOfLake(String commitUser) {
     }
 
     @Override
-    public void close() throws Exception {}
+    public void close() throws Exception {
+        allocator.close();
+    }
 }

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

@@ -50,7 +50,7 @@ public LanceLakeWriter(Configuration options, WriterInitContext writerInitContex
                         writerInitContext.tablePath().getTableName());
         int batchSize = LanceConfig.getBatchSize(config);
         Optional<Schema> schema = LanceDatasetAdapter.getSchema(config);
-        if (!schema.isPresent()) {
+        if (schema.isEmpty()) {
             throw new IOException("Fail to get dataset " + config.getDatasetUri() + " in Lance.");
         }
 

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

@@ -36,18 +36,17 @@ public int getVersion() {
 
     @Override
     public byte[] serialize(LanceWriteResult lanceWriteResult) throws IOException {
-        ByteArrayOutputStream baos = new ByteArrayOutputStream();
-        ObjectOutputStream oos = new ObjectOutputStream(baos);
-        oos.writeObject(lanceWriteResult);
-        oos.close();
-        return baos.toByteArray();
+        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                ObjectOutputStream oos = new ObjectOutputStream(baos); ) {
+            oos.writeObject(lanceWriteResult);
+            return baos.toByteArray();
+        }
     }
 
     @Override
     public LanceWriteResult deserialize(int version, byte[] serialized) throws IOException {
-        ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
-        ObjectInputStream ois = new ObjectInputStream(bais);
-        try {
+        try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+                ObjectInputStream ois = new ObjectInputStream(bais)) {
             return (LanceWriteResult) ois.readObject();
         } catch (ClassNotFoundException e) {
             throw new IOException("Couldn't deserialize LanceWriteResult", e);

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

@@ -24,7 +24,6 @@
 import com.lancedb.lance.Dataset;
 import com.lancedb.lance.Fragment;
 import com.lancedb.lance.FragmentMetadata;
-import com.lancedb.lance.FragmentOperation;
 import com.lancedb.lance.ReadOptions;
 import com.lancedb.lance.Transaction;
 import com.lancedb.lance.WriteParams;
@@ -59,25 +58,6 @@ public static Optional<Schema> getSchema(LanceConfig config) {
         }
     }
 
-    public static long appendFragments(LanceConfig config, List<FragmentMetadata> fragments) {
-        FragmentOperation.Append appendOp = new FragmentOperation.Append(fragments);
-        String uri = config.getDatasetUri();
-        ReadOptions options = LanceConfig.genReadOptionFromConfig(config);
-        try (Dataset datasetRead = Dataset.open(allocator, uri, options)) {
-            Dataset datasetWrite =
-                    Dataset.commit(
-                            allocator,
-                            config.getDatasetUri(),
-                            appendOp,
-                            java.util.Optional.of(datasetRead.version()),
-                            options.getStorageOptions());
-            long version = datasetWrite.version();
-            datasetWrite.close();
-            // Dataset.create returns version 1
-            return version - 1;
-        }
-    }
-
     public static long commitAppend(
             LanceConfig config, List<FragmentMetadata> fragments, Map<String, String> properties) {
         String uri = config.getDatasetUri();

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

@@ -50,7 +50,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** IT case for tiering tables to lance. */
-public class LanceTieringITCase extends FlinkLanceTieringTestBase {
+class LanceTieringITCase extends FlinkLanceTieringTestBase {
     protected static final String DEFAULT_DB = "fluss";
     private static StreamExecutionEnvironment execEnv;
     private static Configuration lanceConf;
@@ -124,7 +124,6 @@ private void checkSnapshotPropertyInLance(
 
     private void checkDataInLanceAppendOnlyTable(LanceConfig config, List<InternalRow> expectedRows)
             throws Exception {
-
         try (Dataset dataset =
                 Dataset.open(
                         allocator,

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

@@ -65,7 +65,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** The UT for tiering to Lance via {@link LanceLakeTieringFactory}. */
-public class LanceTieringTest {
+class LanceTieringTest {
     private @TempDir File tempWarehouseDir;
     private LanceLakeTieringFactory lanceLakeTieringFactory;
     private Configuration configuration;