use GenericTaskDeltaWriter

Author: xx789633

fluss-lake/fluss-lake-iceberg/src/main/java/com/alibaba/fluss/lake/iceberg/tiering/BaseDeltaTaskWriter.java

@@ -17,6 +17,8 @@
 
 package com.alibaba.fluss.lake.iceberg.tiering;
 
+import com.alibaba.fluss.lake.iceberg.tiering.writer.AppendOnlyWriter;
+import com.alibaba.fluss.lake.iceberg.tiering.writer.DeltaTaskWriter;
 import com.alibaba.fluss.lake.writer.LakeWriter;
 import com.alibaba.fluss.lake.writer.WriterInitContext;
 import com.alibaba.fluss.metadata.TablePath;
@@ -29,16 +31,13 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.Catalog;
-import org.apache.iceberg.io.FileAppenderFactory;
-import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.OutputFileFactory;
-import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.io.UnpartitionedWriter;
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.util.PropertyUtil;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 
 import static com.alibaba.fluss.lake.iceberg.utils.IcebergConversions.toIceberg;
@@ -50,7 +49,7 @@ public class IcebergLakeWriter implements LakeWriter<IcebergWriteResult> {
 
     private final Catalog icebergCatalog;
     private final Table icebergTable;
-    private final TaskWriter recordWriter;
+    private final RecordWriter recordWriter;
 
     public IcebergLakeWriter(
             IcebergCatalogProvider icebergCatalogProvider, WriterInitContext writerInitContext)
@@ -63,13 +62,14 @@ public IcebergLakeWriter(
         this.recordWriter = createRecordWriter(writerInitContext);
     }
 
-    private TaskWriter<LogRecord> createRecordWriter(WriterInitContext writerInitContext) {
+    private RecordWriter createRecordWriter(WriterInitContext writerInitContext) {
         Schema schema = icebergTable.schema();
         List<Integer> equalityFieldIds = Lists.newArrayList(schema.identifierFieldIds());
         PartitionSpec spec = icebergTable.spec();
-        FileFormat format = fileFormat(icebergTable);
 
-        FileAppenderFactory<LogRecord> appenderFactory = new FlussAppenderFactory();
+        // Get target file size from table properties
+        long targetFileSize = targetFileSize(icebergTable);
+        FileFormat format = fileFormat(icebergTable);
         OutputFileFactory outputFileFactory =
                 OutputFileFactory.builderFor(
                                 icebergTable,
@@ -78,28 +78,30 @@ private TaskWriter<LogRecord> createRecordWriter(WriterInitContext writerInitCon
                                 0)
                         .format(format)
                         .build();
-        // Get target file size from table properties
-        long targetFileSizeBytes = targetFileSize(icebergTable);
-        FileIO fileIO = icebergTable.io();
 
         if (equalityFieldIds == null || equalityFieldIds.isEmpty()) {
             if (spec.isUnpartitioned()) {
+                return new AppendOnlyWriter(
+                        icebergTable,
+                        writerInitContext.schema().getRowType(),
+                        writerInitContext.tableBucket(),
+                        null, // No partition for non-partitioned table
+                        Collections.emptyList(), // No partition keys
+                        format,
+                        outputFileFactory,
+                        targetFileSize);
             }
         } else {
             if (spec.isUnpartitioned()) {
-                return new UnpartitionedDeltaWriter(
-                        spec,
+                return new DeltaTaskWriter(
+                        icebergTable,
+                        writerInitContext.schema().getRowType(),
+                        writerInitContext.tableBucket(),
+                        null, // No partition for non-partitioned table
+                        Collections.emptyList(), // No partition keys);
                         format,
-                        appenderFactory,
                         outputFileFactory,
-                        fileIO,
-                        targetFileSizeBytes,
-                        schema,
-                        writerInitContext.schema().getRowType(),
-                        equalityFieldIds,
-                        writerInitContext.tableBucket().getBucket());
-            } else {
-
+                        targetFileSize);
             }
         }
         return null;