nit

xx789633 · xx789633 · commit 288f050de0c0 · 2025-09-04T23:16:02.000+08:00
diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java
@@ -61,6 +61,7 @@ public class IcebergRewriteDataFiles {
 
     private static final Logger LOG = LoggerFactory.getLogger(IcebergRewriteDataFiles.class);
 
+    // TODO: get this value from table properties
     private static final int MIN_FILES_TO_COMPACT = 3;
 
     private final Table table;
diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java
@@ -56,12 +56,9 @@ public class IcebergLakeWriter implements LakeWriter<IcebergWriteResult> {
 
     protected static final Logger LOG = LoggerFactory.getLogger(IcebergLakeWriter.class);
 
-    private static final String AUTO_MAINTENANCE_KEY = "table.datalake.auto-maintenance";
-
     private final Catalog icebergCatalog;
     private final Table icebergTable;
     private final RecordWriter recordWriter;
-    private final boolean autoMaintenanceEnabled;
 
     @Nullable private final ExecutorService compactionExecutor;
     @Nullable private CompletableFuture<RewriteDataFileResult> compactionFuture;
@@ -72,15 +69,10 @@ public IcebergLakeWriter(
         this.icebergCatalog = icebergCatalogProvider.get();
         this.icebergTable = getTable(writerInitContext.tablePath());
 
-        // Check auto-maintenance from table properties
-        this.autoMaintenanceEnabled =
-                Boolean.parseBoolean(
-                        icebergTable.properties().getOrDefault(AUTO_MAINTENANCE_KEY, "false"));
-
         // Create a record writer
         this.recordWriter = createRecordWriter(writerInitContext);
 
-        if (autoMaintenanceEnabled) {
+        if (writerInitContext.tableInfo().getTableConfig().isDataLakeAutoCompaction()) {
             this.compactionExecutor =
                     Executors.newSingleThreadExecutor(
                             new ExecutorThreadFactory(
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java
@@ -18,60 +18,26 @@
 
 package org.apache.fluss.lake.iceberg.maintenance;
 
-import org.apache.flink.core.execution.JobClient;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.fluss.config.Configuration;
 import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase;
-import org.apache.fluss.lake.iceberg.tiering.IcebergCatalogProvider;
-import org.apache.fluss.lake.iceberg.tiering.writer.TaskWriterFactory;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TablePath;
-
 import org.apache.fluss.row.BinaryString;
 import org.apache.fluss.row.Decimal;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.TimestampLtz;
 import org.apache.fluss.row.TimestampNtz;
 import org.apache.fluss.types.DataTypes;
 import org.apache.fluss.utils.TypeUtils;
-import org.apache.iceberg.AppendFiles;
-import org.apache.iceberg.DataFile;
-import org.apache.iceberg.FileScanTask;
-import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.RewriteFiles;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.Table;
-import org.apache.iceberg.catalog.Catalog;
-import org.apache.iceberg.catalog.Namespace;
-import org.apache.iceberg.catalog.SupportsNamespaces;
-import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.data.IcebergGenerics;
-import org.apache.iceberg.data.Record;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.types.Types;
+
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
-import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg;
-import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
-import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
-import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
 import static org.apache.fluss.testutils.DataTestUtils.row;
-import static org.apache.fluss.utils.Preconditions.checkState;
-import static org.assertj.core.api.Assertions.assertThat;
 
 /** Integration test to for Iceberg compaction. */
 class IcebergRewriteITCase extends FlinkIcebergTieringTestBase {
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java
@@ -250,6 +250,7 @@ protected long createPkTable(TablePath tablePath, int bucketNum) throws Exceptio
                         .distributedBy(bucketNum)
                         .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")
                         .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500))
+                        .property(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(), "true")
                         .build();
         return createTable(tablePath, table1Descriptor);
     }
@@ -386,7 +387,8 @@ protected void checkDataInIcebergAppendOnlyTable(
         }
     }
 
-    protected void checkFileInIcebergTable(TablePath tablePath, int expectedFileCount) throws IOException {
+    protected void checkFileInIcebergTable(TablePath tablePath, int expectedFileCount)
+            throws IOException {
         org.apache.iceberg.Table table = icebergCatalog.loadTable(toIceberg(tablePath));
         int count = 0;
         try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/resources/log4j2-test.properties b/fluss-lake/fluss-lake-iceberg/src/test/resources/log4j2-test.properties
@@ -17,7 +17,7 @@
 #
 # Set root logger level to OFF to not flood build logs
 # set manually to INFO for debugging purposes
-rootLogger.level=OFF
+rootLogger.level=INFO
 rootLogger.appenderRef.test.ref=TestLogger
 appender.testlogger.name=TestLogger
 appender.testlogger.type=CONSOLE

Original file line number	Diff line number	Diff line change
`@@ -250,6 +250,7 @@ protected long createPkTable(TablePath tablePath, int bucketNum) throws Exceptio`
`250`	`250`	`.distributedBy(bucketNum)`
`251`	`251`	`.property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")`
`252`	`252`	`.property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500))`
	`253`	`+ .property(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(), "true")`
`253`	`254`	`.build();`
`254`	`255`	`return createTable(tablePath, table1Descriptor);`
`255`	`256`	`}`
`@@ -386,7 +387,8 @@ protected void checkDataInIcebergAppendOnlyTable(`
`386`	`387`	`}`
`387`	`388`	`}`
`388`	`389`
`389`		`- protected void checkFileInIcebergTable(TablePath tablePath, int expectedFileCount) throws IOException {`
	`390`	`+ protected void checkFileInIcebergTable(TablePath tablePath, int expectedFileCount)`
	`391`	`+ throws IOException {`
`390`	`392`	`org.apache.iceberg.Table table = icebergCatalog.loadTable(toIceberg(tablePath));`
`391`	`393`	`int count = 0;`
`392`	`394`	`try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {`
Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@`
`17`	`17`	`#`
`18`	`18`	`# Set root logger level to OFF to not flood build logs`
`19`	`19`	`# set manually to INFO for debugging purposes`
`20`		`-rootLogger.level=OFF`
	`20`	`+rootLogger.level=INFO`
`21`	`21`	`rootLogger.appenderRef.test.ref=TestLogger`
`22`	`22`	`appender.testlogger.name=TestLogger`
`23`	`23`	`appender.testlogger.type=CONSOLE`