[core] Enables buffer spill when targetFileSize is greater than the write buffer size.

Author: LinMingQiang

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1947,9 +1947,14 @@ public long writeBufferSize() {
         return options.get(WRITE_BUFFER_SIZE).getBytes();
     }
 
-    public boolean writeBufferSpillable(boolean usingObjectStore, boolean isStreaming) {
+    public boolean writeBufferSpillable(
+            boolean usingObjectStore, boolean isStreaming, boolean hasPrimaryKey) {
         // if not streaming mode, we turn spillable on by default.
-        return options.getOptional(WRITE_BUFFER_SPILLABLE).orElse(usingObjectStore || !isStreaming);
+        return options.getOptional(WRITE_BUFFER_SPILLABLE)
+                .orElse(
+                        usingObjectStore
+                                || !isStreaming
+                                || targetFileSize(hasPrimaryKey) > writeBufferSize());
     }
 
     public MemorySize writeBufferSpillDiskSize() {

paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java

@@ -125,7 +125,7 @@ protected RecordWriter<InternalRow> createWriter(
                 pathFactory.createDataFilePathFactory(partition, bucket),
                 restoreIncrement,
                 options.useWriteBufferForAppend() || forceBufferSpill,
-                options.writeBufferSpillable(fileIO.isObjectStore(), isStreamingMode)
+                options.writeBufferSpillable(fileIO.isObjectStore(), isStreamingMode, false)
                         || forceBufferSpill,
                 options.fileCompression(),
                 options.spillCompressOptions(),

paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java

@@ -229,7 +229,7 @@ protected MergeTreeWriter createWriter(
 
     @VisibleForTesting
     public boolean bufferSpillable() {
-        return options.writeBufferSpillable(fileIO.isObjectStore(), isStreamingMode);
+        return options.writeBufferSpillable(fileIO.isObjectStore(), isStreamingMode, true);
     }
 
     private CompactManager createCompactManager(

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterOperatorTest.java

@@ -87,6 +87,7 @@ public void testPrimaryKeyTableMetrics() throws Exception {
         Options options = new Options();
         options.set("bucket", "1");
         options.set("write-buffer-size", "256 b");
+        options.set("write-buffer-spillable", "false");
         options.set("page-size", "32 b");
 
         FileStoreTable table =
@@ -332,6 +333,7 @@ public void testNumWritersMetric() throws Exception {
         Options options = new Options();
         options.set("bucket", "1");
         options.set("write-buffer-size", "256 b");
+        options.set("write-buffer-spillable", "false");
         options.set("page-size", "32 b");
 
         FileStoreTable fileStoreTable =