fix

Author: JingsongLi

docs/layouts/shortcodes/generated/core_configuration.html

@@ -272,6 +272,12 @@
             <td>Integer</td>
             <td>Percentage flexibility while comparing sorted run size for changelog mode table. If the candidate sorted run(s) size is 1% smaller than the next sorted run's size, then include next sorted run into this candidate set.</td>
         </tr>
+        <tr>
+            <td><h5>compaction.total-size-threshold</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>MemorySize</td>
+            <td>When total size is smaller than this threshold, force a full compaction.</td>
+        </tr>
         <tr>
             <td><h5>consumer-id</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java

@@ -19,7 +19,12 @@
 package org.apache.paimon.flink;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.deletionvectors.DeletionVector;
 import org.apache.paimon.flink.util.AbstractTestBase;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
@@ -40,13 +45,13 @@
 
 import java.math.BigDecimal;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.Collectors;
 
+import static java.util.Collections.singletonList;
 import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -56,7 +61,56 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
 
     @Override
     protected List<String> ddl() {
-        return Collections.singletonList("CREATE TABLE IF NOT EXISTS T (a INT, b INT, c INT)");
+        return singletonList("CREATE TABLE IF NOT EXISTS T (a INT, b INT, c INT)");
+    }
+
+    @Test
+    public void testFullCompactionNoDv() throws Catalog.TableNotExistException {
+        sql(
+                "CREATE TEMPORARY TABLE GEN (a INT) WITH ("
+                        + "'connector'='datagen', "
+                        + "'number-of-rows'='1000', "
+                        + "'fields.a.kind'='sequence', "
+                        + "'fields.a.start'='0', "
+                        + "'fields.a.end'='1000')");
+        sql(
+                "CREATE TABLE T1 (a INT PRIMARY KEY NOT ENFORCED, b STRING, c STRING) WITH ("
+                        + "'bucket' = '1', "
+                        + "'file.format' = 'avro', "
+                        + "'file.compression' = 'null', "
+                        + "'deletion-vectors.enabled' = 'true')");
+        batchSql("INSERT INTO T1 SELECT a, 'unknown', 'unknown' FROM GEN");
+
+        // first insert, producing dv files
+        batchSql("INSERT INTO T1 VALUES (1, '22', '33')");
+        FileStoreTable table = paimonTable("T1");
+        Snapshot snapshot = table.latestSnapshot().get();
+        assertThat(deletionVectors(table, snapshot)).hasSize(1);
+        assertThat(batchSql("SELECT * FROM T1 WHERE a = 1")).containsExactly(Row.of(1, "22", "33"));
+
+        // second insert, producing no dv files
+        batchSql("ALTER TABLE T1 SET ('compaction.total-size-threshold' = '1m')");
+        batchSql("INSERT INTO T1 VALUES (1, '44', '55')");
+        snapshot = table.latestSnapshot().get();
+        assertThat(deletionVectors(table, snapshot)).hasSize(0);
+        assertThat(batchSql("SELECT * FROM T1 WHERE a = 1")).containsExactly(Row.of(1, "44", "55"));
+
+        // third insert, producing no dv files, same index manifest
+        batchSql("INSERT INTO T1 VALUES (1, '66', '77')");
+        assertThat(table.latestSnapshot().get().indexManifest())
+                .isEqualTo(snapshot.indexManifest());
+        assertThat(batchSql("SELECT * FROM T1 WHERE a = 1")).containsExactly(Row.of(1, "66", "77"));
+    }
+
+    private Map<String, DeletionVector> deletionVectors(FileStoreTable table, Snapshot snapshot) {
+        assertThat(snapshot.indexManifest()).isNotNull();
+        List<IndexManifestEntry> indexManifestEntries =
+                table.indexManifestFileReader().read(snapshot.indexManifest());
+        assertThat(indexManifestEntries.size()).isEqualTo(1);
+        IndexFileMeta indexFileMeta = indexManifestEntries.get(0).indexFile();
+        return table.store()
+                .newIndexFileHandler()
+                .readAllDeletionVectors(singletonList(indexFileMeta));
     }
 
     @Test
@@ -794,7 +848,7 @@ public void testIncrementScanMode() throws Exception {
         // snapshot 5,6
         String dataId =
                 TestValuesTableFactory.registerData(
-                        Collections.singletonList(Row.ofKind(RowKind.DELETE, 2, "B")));
+                        singletonList(Row.ofKind(RowKind.DELETE, 2, "B")));
         sEnv.executeSql(
                 "CREATE TEMPORARY TABLE source (id INT, v STRING) "
                         + "WITH ('connector' = 'values', 'bounded' = 'true', 'data-id' = '"