nit

Author: xx789633

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java

@@ -21,20 +21,13 @@
 import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase;
 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.flink.core.execution.JobClient;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
-import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -55,29 +48,29 @@ protected static void beforeAll() {
     }
 
     @Test
-    void testCompaction() throws Exception {
+    void testLogTableCompaction() throws Exception {
         JobClient jobClient = buildTieringJob(execEnv);
         try {
-            TablePath t1 = TablePath.of(DEFAULT_DB, "pkTable");
-            long t1Id = createPkTable(t1, true);
+            TablePath t1 = TablePath.of(DEFAULT_DB, "logTable");
+            long t1Id = createLogTable(t1, true);
             TableBucket t1Bucket = new TableBucket(t1Id, 0);
 
-            writeFullTypeRows(t1, t1Bucket, 1, 3, 3);
+            writeTableRecords(t1, t1Bucket, 1, 3, 3);
 
-            writeFullTypeRows(t1, t1Bucket, 4, 6, 6);
+            writeTableRecords(t1, t1Bucket, 4, 6, 6);
 
-            writeFullTypeRows(t1, t1Bucket, 7, 9, 9);
+            writeTableRecords(t1, t1Bucket, 7, 9, 9);
             checkFileCountInIcebergTable(t1, 3);
 
             // trigger compaction
-            writeFullTypeRows(t1, t1Bucket, 10, 12, 12);
+            writeTableRecords(t1, t1Bucket, 10, 12, 12);
             checkFileCountInIcebergTable(t1, 2);
         } finally {
             jobClient.cancel().get();
         }
     }
 
-    private void writeFullTypeRows(
+    private void writeTableRecords(
             TablePath tablePath,
             TableBucket tableBucket,
             int from,
@@ -86,29 +79,9 @@ private void writeFullTypeRows(
             throws Exception {
         List<InternalRow> rows = new ArrayList<>();
         for (int i = from; i <= to; i++) {
-            rows.add(
-                    row(
-                            true,
-                            (byte) 100,
-                            (short) 200,
-                            i,
-                            i + 400L,
-                            500.1f,
-                            600.0d,
-                            "v1",
-                            Decimal.fromUnscaledLong(900, 5, 2),
-                            Decimal.fromBigDecimal(new BigDecimal(1000), 20, 0),
-                            TimestampLtz.fromEpochMillis(1698235273400L),
-                            TimestampLtz.fromEpochMillis(1698235273400L, 7000),
-                            TimestampNtz.fromMillis(1698235273501L),
-                            TimestampNtz.fromMillis(1698235273501L, 8000),
-                            new byte[] {5, 6, 7, 8},
-                            TypeUtils.castFromString("2023-10-25", DataTypes.DATE()),
-                            TypeUtils.castFromString("09:30:00.0", DataTypes.TIME()),
-                            BinaryString.fromString("abc"),
-                            new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}));
+            rows.add(row(i, "v" + i));
         }
-        writeRows(tablePath, rows, false);
+        writeRows(tablePath, rows, true);
         assertReplicaStatus(tableBucket, expectedLogEndOffset);
     }
 }

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java

@@ -197,14 +197,16 @@ protected long createPkTable(TablePath tablePath, boolean enableAutoCompaction)
     }
 
     protected long createLogTable(TablePath tablePath) throws Exception {
-        return createLogTable(tablePath, 1);
+        return createLogTable(tablePath, false);
     }
 
-    protected long createLogTable(TablePath tablePath, int bucketNum) throws Exception {
-        return createLogTable(tablePath, bucketNum, false);
+    protected long createLogTable(TablePath tablePath, boolean enableAutoCompaction)
+            throws Exception {
+        return createLogTable(tablePath, 1, false, enableAutoCompaction);
     }
 
-    protected long createLogTable(TablePath tablePath, int bucketNum, boolean isPartitioned)
+    protected long createLogTable(
+            TablePath tablePath, int bucketNum, boolean isPartitioned, boolean enableAutoCompaction)
             throws Exception {
         Schema.Builder schemaBuilder =
                 Schema.newBuilder().column("a", DataTypes.INT()).column("b", DataTypes.STRING());
@@ -222,6 +224,9 @@ protected long createLogTable(TablePath tablePath, int bucketNum, boolean isPart
             tableBuilder.property(
                     ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT, AutoPartitionTimeUnit.YEAR);
         }
+        if (enableAutoCompaction) {
+            tableBuilder.property(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(), "true");
+        }
         tableBuilder.schema(schemaBuilder.build());
         return createTable(tablePath, tableBuilder.build());
     }