nit

Author: xx789633

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

@@ -62,7 +62,7 @@ public class PaimonLakeCatalog implements LakeCatalog {
     private final Catalog paimonCatalog;
 
     // for fluss config
-    private static final String FLUSS_CONF_PREFIX = "fluss.";
+    public static final String FLUSS_CONF_PREFIX = "fluss.";
     // for paimon config
     private static final String PAIMON_CONF_PREFIX = "paimon.";
 

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java

@@ -24,13 +24,19 @@
 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 java.util.Objects;
 
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION;
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.FLUSS_CONF_PREFIX;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
 
 /** Implementation of {@link LakeWriter} for Paimon. */
@@ -97,7 +103,21 @@ public void close() throws IOException {
 
     private FileStoreTable getTable(TablePath tablePath) 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(),
+                            Objects.equals(
+                                            table.schema()
+                                                    .options()
+                                                    .get(
+                                                            FLUSS_CONF_PREFIX
+                                                                    + TABLE_DATALAKE_AUTO_COMPACTION
+                                                                            .key()),
+                                            "true")
+                                    ? Boolean.FALSE.toString()
+                                    : Boolean.TRUE.toString());
+            return table.copy(compactionOptions);
         } catch (Exception e) {
             throw new IOException("Failed to get table " + tablePath + " in Paimon.", e);
         }