implement bytes written tracking in LanceLakeWriter

Author: beryllw

fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/tiering/LanceLakeWriter.java

@@ -29,6 +29,7 @@
 import com.lancedb.lance.Fragment;
 import com.lancedb.lance.FragmentMetadata;
 import com.lancedb.lance.WriteParams;
+import com.lancedb.lance.fragment.DataFile;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.vector.VectorSchemaRoot;
@@ -52,6 +53,7 @@ public class LanceLakeWriter implements LakeWriter<LanceWriteResult> {
 
     private final ShadedArrowBatchWriter arrowWriter;
     private final List<FragmentMetadata> allFragments;
+    private long bytesWritten = -1L;
 
     public LanceLakeWriter(Configuration options, WriterInitContext writerInitContext)
             throws IOException {
@@ -117,13 +119,32 @@ private List<FragmentMetadata> flush() throws IOException {
         }
     }
 
+    private long computeBytesWritten(List<FragmentMetadata> fragments) {
+        long total = 0L;
+        for (FragmentMetadata fragment : fragments) {
+            for (DataFile dataFile : fragment.getFiles()) {
+                Long size = dataFile.getFileSizeBytes();
+                if (size != null) {
+                    total += size;
+                }
+            }
+        }
+        return total;
+    }
+
     @Override
     public LanceWriteResult complete() throws IOException {
         List<FragmentMetadata> fragments = flush();
         allFragments.addAll(fragments);
+        bytesWritten = computeBytesWritten(allFragments);
         return new LanceWriteResult(allFragments);
     }
 
+    @Override
+    public long getBytesWritten() {
+        return bytesWritten;
+    }
+
     @Override
     public void close() throws IOException {
         if (arrowWriter != null) {

fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java

@@ -160,7 +160,7 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
                     // serialize/deserialize writeResult
                     LanceWriteResult lanceWriteResult = lakeWriter.complete();
                     // Verify that getBytesWritten returns a positive value after complete
-                    assertThat(lakeWriter.getBytesWritten()).isEqualTo(-1L);
+                    assertThat(lakeWriter.getBytesWritten()).isGreaterThan(0L);
                     byte[] serialized = writeResultSerializer.serialize(lanceWriteResult);
                     lanceWriteResults.add(
                             writeResultSerializer.deserialize(