[lake] Introduce table.datalake.auto-compaction options (#1612)

xx789 · polyzos · commit 7a841a30b99b · 2025-08-31T17:27:39.000+03:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1290,6 +1290,13 @@ public class ConfigOptions {
                                     + "Based on this target freshness, the Fluss service automatically moves data from the Fluss table and updates to the datalake table, so that the data in the datalake table is kept up to date within this target. "
                                     + "If the data does not need to be as fresh, you can specify a longer target freshness time to reduce costs.");
 
+    public static final ConfigOption<Boolean> TABLE_DATALAKE_AUTO_COMPACTION =
+            key("table.datalake.auto-compaction")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, compaction will be triggered automatically when tiering service writes to the datalake. It is disabled by default.");
+
     public static final ConfigOption<MergeEngineType> TABLE_MERGE_ENGINE =
             key("table.merge-engine")
                     .enumType(MergeEngineType.class)
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
@@ -94,6 +94,11 @@ public Duration getDataLakeFreshness() {
         return config.get(ConfigOptions.TABLE_DATALAKE_FRESHNESS);
     }
 
+    /** Whether auto compaction is enabled. */
+    public boolean isDataLakeAutoCompaction() {
+        return config.get(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION);
+    }
+
     /** Gets the optional merge engine type of the table. */
     public Optional<MergeEngineType> getMergeEngineType() {
         return config.getOptional(ConfigOptions.TABLE_MERGE_ENGINE);
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java b/fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java
@@ -18,17 +18,16 @@
 package org.apache.fluss.lake.writer;
 
 import org.apache.fluss.annotation.PublicEvolving;
-import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 
 import javax.annotation.Nullable;
 
-import java.util.Map;
-
 /**
  * The WriterInitContext interface provides the context needed to create a LakeWriter. It includes
- * methods to obtain the table path, table bucket, and an optional partition.
+ * methods to obtain the table path, table bucket, an optional partition, and table metadata
+ * information.
  *
  * @since 0.7
  */
@@ -58,16 +57,9 @@ public interface WriterInitContext {
     String partition();
 
     /**
-     * Returns the table schema.
-     *
-     * @return the table schema
-     */
-    Schema schema();
-
-    /**
-     * Returns the table custom properties.
+     * Returns the Fluss table info.
      *
-     * @return the table custom properties
+     * @return the Fluss table info
      */
-    Map<String, String> customProperties();
+    TableInfo tableInfo();
 }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
@@ -309,8 +309,7 @@ private LakeWriter<WriteResult> getOrCreateLakeWriter(
                                     currentTablePath,
                                     bucket,
                                     partitionName,
-                                    currentTable.getTableInfo().getSchema(),
-                                    currentTable.getTableInfo().getCustomProperties().toMap()));
+                                    currentTable.getTableInfo()));
             lakeWriters.put(bucket, lakeWriter);
         }
         return lakeWriter;
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java
@@ -18,34 +18,29 @@
 package org.apache.fluss.flink.tiering.source;
 
 import org.apache.fluss.lake.writer.WriterInitContext;
-import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 
 import javax.annotation.Nullable;
 
-import java.util.Map;
-
 /** The implementation of {@link WriterInitContext}. */
 public class TieringWriterInitContext implements WriterInitContext {
 
     private final TablePath tablePath;
     private final TableBucket tableBucket;
-    private final Schema schema;
     @Nullable private final String partition;
-    private final Map<String, String> customProperties;
+    private final TableInfo tableInfo;
 
     public TieringWriterInitContext(
             TablePath tablePath,
             TableBucket tableBucket,
             @Nullable String partition,
-            Schema schema,
-            Map<String, String> customProperties) {
+            TableInfo tableInfo) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.partition = partition;
-        this.schema = schema;
-        this.customProperties = customProperties;
+        this.tableInfo = tableInfo;
     }
 
     @Override
@@ -65,12 +60,7 @@ public String partition() {
     }
 
     @Override
-    public Schema schema() {
-        return schema;
-    }
-
-    @Override
-    public Map<String, String> customProperties() {
-        return customProperties;
+    public TableInfo tableInfo() {
+        return tableInfo;
     }
 }
diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/writer/AppendOnlyTaskWriter.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/writer/AppendOnlyTaskWriter.java
@@ -35,7 +35,7 @@ public AppendOnlyTaskWriter(
         super(
                 taskWriter,
                 icebergTable.schema(),
-                writerInitContext.schema().getRowType(),
+                writerInitContext.tableInfo().getRowType(),
                 writerInitContext.tableBucket());
     }
 
diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/writer/DeltaTaskWriter.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/writer/DeltaTaskWriter.java
@@ -35,7 +35,7 @@ public DeltaTaskWriter(
         super(
                 taskWriter,
                 icebergTable.schema(),
-                writerInitContext.schema().getRowType(),
+                writerInitContext.tableInfo().getRowType(),
                 writerInitContext.tableBucket());
     }
 
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java
@@ -17,12 +17,15 @@
 
 package org.apache.fluss.lake.iceberg.tiering;
 
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.lake.committer.LakeCommitter;
 import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
 import org.apache.fluss.lake.writer.LakeWriter;
 import org.apache.fluss.lake.writer.WriterInitContext;
 import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.record.ChangeType;
 import org.apache.fluss.record.GenericRecord;
@@ -119,6 +122,19 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
                                 isPartitionedTable ? "partitioned" : "unpartitioned"));
         createTable(tablePath, isPrimaryKeyTable, isPartitionedTable);
 
+        TableDescriptor descriptor =
+                TableDescriptor.builder()
+                        .schema(
+                                org.apache.fluss.metadata.Schema.newBuilder()
+                                        .column("c1", DataTypes.INT())
+                                        .column("c2", DataTypes.STRING())
+                                        .column("c3", DataTypes.STRING())
+                                        .build())
+                        .distributedBy(BUCKET_NUM)
+                        .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+                        .build();
+        TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
         Table icebergTable = icebergCatalog.loadTable(toIceberg(tablePath));
 
         Map<Tuple2<String, Integer>, List<LogRecord>> recordsByBucket = new HashMap<>();
@@ -144,7 +160,7 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
             for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
                 String partition = entry.getValue();
                 try (LakeWriter<IcebergWriteResult> writer =
-                        createLakeWriter(tablePath, bucket, partition, entry.getKey())) {
+                        createLakeWriter(tablePath, bucket, partition, entry.getKey(), tableInfo)) {
                     Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
                     Tuple2<List<LogRecord>, List<LogRecord>> writeAndExpectRecords =
                             isPrimaryKeyTable
@@ -198,7 +214,11 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
     }
 
     private LakeWriter<IcebergWriteResult> createLakeWriter(
-            TablePath tablePath, int bucket, @Nullable String partition, @Nullable Long partitionId)
+            TablePath tablePath,
+            int bucket,
+            @Nullable String partition,
+            @Nullable Long partitionId,
+            TableInfo tableInfo)
             throws IOException {
         return icebergLakeTieringFactory.createLakeWriter(
                 new WriterInitContext() {
@@ -219,17 +239,8 @@ public String partition() {
                     }
 
                     @Override
-                    public Map<String, String> customProperties() {
-                        return Collections.emptyMap();
-                    }
-
-                    @Override
-                    public org.apache.fluss.metadata.Schema schema() {
-                        return org.apache.fluss.metadata.Schema.newBuilder()
-                                .column("c1", DataTypes.INT())
-                                .column("c2", DataTypes.STRING())
-                                .column("c3", DataTypes.STRING())
-                                .build();
+                    public TableInfo tableInfo() {
+                        return tableInfo;
                     }
                 });
     }
diff --git a/fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/tiering/LanceLakeWriter.java b/fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/tiering/LanceLakeWriter.java
@@ -45,7 +45,7 @@ public LanceLakeWriter(Configuration options, WriterInitContext writerInitContex
         LanceConfig config =
                 LanceConfig.from(
                         options.toMap(),
-                        writerInitContext.customProperties(),
+                        writerInitContext.tableInfo().getCustomProperties().toMap(),
                         writerInitContext.tablePath().getDatabaseName(),
                         writerInitContext.tablePath().getTableName());
         int batchSize = LanceConfig.getBatchSize(config);
@@ -56,7 +56,7 @@ public LanceLakeWriter(Configuration options, WriterInitContext writerInitContex
 
         this.arrowWriter =
                 LanceDatasetAdapter.getArrowWriter(
-                        schema.get(), batchSize, writerInitContext.schema().getRowType());
+                        schema.get(), batchSize, writerInitContext.tableInfo().getRowType());
 
         WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
         Callable<List<FragmentMetadata>> fragmentCreator =
diff --git a/fluss-lake/fluss-lake-lance/src/test/java/com/alibaba/fluss/lake/lance/tiering/LanceTieringTest.java b/fluss-lake/fluss-lake-lance/src/test/java/com/alibaba/fluss/lake/lance/tiering/LanceTieringTest.java
@@ -17,6 +17,7 @@
 
 package org.apache.fluss.lake.lance.tiering;
 
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.lake.committer.CommittedLakeSnapshot;
 import org.apache.fluss.lake.committer.LakeCommitter;
@@ -28,6 +29,8 @@
 import org.apache.fluss.lake.writer.WriterInitContext;
 import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.record.ChangeType;
 import org.apache.fluss.record.GenericRecord;
@@ -96,6 +99,15 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
                         tablePath.getTableName());
         Schema schema = createTable(config);
 
+        TableDescriptor descriptor =
+                TableDescriptor.builder()
+                        .schema(schema)
+                        .distributedBy(bucketNum)
+                        .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+                        .customProperties(customProperties)
+                        .build();
+        TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
         List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
         SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
                 lanceLakeTieringFactory.getWriteResultSerializer();
@@ -126,7 +138,7 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
             for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
                 String partition = entry.getValue();
                 try (LakeWriter<LanceWriteResult> lakeWriter =
-                        createLakeWriter(tablePath, bucket, partition, schema, customProperties)) {
+                        createLakeWriter(tablePath, bucket, partition, tableInfo)) {
                     Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
                     Tuple2<List<LogRecord>, List<LogRecord>> writeAndExpectRecords =
                             genLogTableRecords(partition, bucket, 10);
@@ -239,11 +251,7 @@ private LakeCommitter<LanceWriteResult, LanceCommittable> createLakeCommitter(
     }
 
     private LakeWriter<LanceWriteResult> createLakeWriter(
-            TablePath tablePath,
-            int bucket,
-            @Nullable String partition,
-            Schema schema,
-            Map<String, String> customProperties)
+            TablePath tablePath, int bucket, @Nullable String partition, TableInfo tableInfo)
             throws IOException {
         return lanceLakeTieringFactory.createLakeWriter(
                 new WriterInitContext() {
@@ -265,13 +273,8 @@ public String partition() {
                     }
 
                     @Override
-                    public org.apache.fluss.metadata.Schema schema() {
-                        return schema;
-                    }
-
-                    @Override
-                    public Map<String, String> customProperties() {
-                        return customProperties;
+                    public TableInfo tableInfo() {
+                        return tableInfo;
                     }
                 });
     }
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java
@@ -24,12 +24,15 @@
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.record.LogRecord;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.CommitMessage;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
 
@@ -43,7 +46,10 @@ public PaimonLakeWriter(
             PaimonCatalogProvider paimonCatalogProvider, WriterInitContext writerInitContext)
             throws IOException {
         this.paimonCatalog = paimonCatalogProvider.get();
-        FileStoreTable fileStoreTable = getTable(writerInitContext.tablePath());
+        FileStoreTable fileStoreTable =
+                getTable(
+                        writerInitContext.tablePath(),
+                        writerInitContext.tableInfo().getTableConfig().isDataLakeAutoCompaction());
 
         List<String> partitionKeys = fileStoreTable.partitionKeys();
 
@@ -95,9 +101,15 @@ public void close() throws IOException {
         }
     }
 
-    private FileStoreTable getTable(TablePath tablePath) throws IOException {
+    private FileStoreTable getTable(TablePath tablePath, boolean isAutoCompaction)
+            throws IOException {
         try {
-            return (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath));
+            FileStoreTable table = (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath));
+            Map<String, String> compactionOptions =
+                    Collections.singletonMap(
+                            CoreOptions.WRITE_ONLY.key(),
+                            isAutoCompaction ? Boolean.FALSE.toString() : Boolean.TRUE.toString());
+            return table.copy(compactionOptions);
         } catch (Exception e) {
             throw new IOException("Failed to get table " + tablePath + " in Paimon.", e);
         }
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
diff --git a/website/docs/maintenance/operations/upgrading.md b/website/docs/maintenance/operations/upgrading.md

Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ public AppendOnlyTaskWriter(`
`35`	`35`	`super(`
`36`	`36`	`taskWriter,`
`37`	`37`	`icebergTable.schema(),`
`38`		`- writerInitContext.schema().getRowType(),`
	`38`	`+ writerInitContext.tableInfo().getRowType(),`
`39`	`39`	`writerInitContext.tableBucket());`
`40`	`40`	`}`
`41`	`41`
Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ public DeltaTaskWriter(`
`35`	`35`	`super(`
`36`	`36`	`taskWriter,`
`37`	`37`	`icebergTable.schema(),`
`38`		`- writerInitContext.schema().getRowType(),`
	`38`	`+ writerInitContext.tableInfo().getRowType(),`
`39`	`39`	`writerInitContext.tableBucket());`
`40`	`40`	`}`
`41`	`41`