diff --git a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFormatWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFormatWriter.java
index 6de063d0d7d4..f6e3ca45cf3d 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFormatWriter.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFormatWriter.java
@@ -104,7 +104,8 @@ private String escapeField(String field) {
 
         // Optimized escaping with early exit checks
         boolean needsQuoting =
-                field.indexOf(csvOptions.fieldDelimiter().charAt(0)) >= 0
+                field.equals(csvOptions.nullLiteral())
+                        || field.indexOf(csvOptions.fieldDelimiter().charAt(0)) >= 0
                         || field.indexOf(csvOptions.lineDelimiter().charAt(0)) >= 0
                         || field.indexOf(quote.charAt(0)) >= 0
                         || (escapable && field.indexOf(escape.charAt(0)) >= 0);
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvParser.java b/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvParser.java
index 8ea5e60904c4..58f7b790c22c 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvParser.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvParser.java
@@ -60,6 +60,7 @@ public class CsvParser {
     private final Mode mode;
     private final StringBuilder buffer;
     private final String[] rowValues;
+    private final boolean[] rowQuoted;
 
     public CsvParser(RowType dataSchemaRowType, int[] projectMapping, CsvOptions options) {
         this.dataSchemaRowType = dataSchemaRowType;
@@ -70,6 +71,7 @@ public CsvParser(RowType dataSchemaRowType, int[] projectMapping, CsvOptions opt
         this.buffer = new StringBuilder(1024);
         int columnCount = Arrays.stream(projectMapping).max().orElse(-1) + 1;
         this.rowValues = new String[columnCount];
+        this.rowQuoted = new boolean[columnCount];
 
         this.separatorChar = options.fieldDelimiter().charAt(0);
         this.quoteChar = options.quoteCharacter().charAt(0);
@@ -91,6 +93,7 @@ public CsvParser(RowType dataSchemaRowType, int[] projectMapping, CsvOptions opt
     @Nullable
     public GenericRow parse(String line) {
         Arrays.fill(rowValues, null);
+        Arrays.fill(rowQuoted, false);
         buffer.setLength(0);
 
         // empty line results in all null values
@@ -136,6 +139,9 @@ public GenericRow parse(String line) {
                             buffer.append(c);
                         }
                     }
+                    if (!inQuotes && buffer.length() == 0) {
+                        rowQuoted[columnIndex] = true;
+                    }
                     inQuotes = !inQuotes;
                 }
                 inField = !inField;
@@ -166,7 +172,7 @@ public GenericRow parse(String line) {
             Exception exception = null;
             String parseValue = rowValues[ordinal];
             try {
-                parseResult = parseField(parseValue, type);
+                parseResult = parseField(parseValue, type, rowQuoted[ordinal]);
             } catch (Exception e) {
                 exception = e;
             }
@@ -203,7 +209,11 @@ private static boolean isAllWhitespace(CharSequence sequence) {
 
     @VisibleForTesting
     public Pair<Boolean, Object> parseField(String field, DataType dataType) {
-        if (field == null || field.equals(nullLiteral)) {
+        return parseField(field, dataType, false);
+    }
+
+    private Pair<Boolean, Object> parseField(String field, DataType dataType, boolean quoted) {
+        if (field == null || (!quoted && field.equals(nullLiteral))) {
             return Pair.of(true, null);
         }
 
diff --git a/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java
index 5e47ebae953e..008bd564ab65 100644
--- a/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java
+++ b/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java
@@ -464,17 +464,17 @@ public void testCsvNullLiteralWriteRead() throws IOException {
 
         String[] nullLiterals = {"", "NULL", "null"};
 
-        // Create test data with null values
-        List<InternalRow> testData =
-                Arrays.asList(
-                        GenericRow.of(1, BinaryString.fromString("Alice"), null),
-                        GenericRow.of(2, null, 100),
-                        GenericRow.of(3, BinaryString.fromString("Charlie"), 300));
-
         for (String nullLiteral : nullLiterals) {
             Options options = new Options();
             options.set(CsvOptions.NULL_LITERAL, nullLiteral);
 
+            List<InternalRow> testData =
+                    Arrays.asList(
+                            GenericRow.of(1, BinaryString.fromString("Alice"), null),
+                            GenericRow.of(2, null, 100),
+                            GenericRow.of(3, BinaryString.fromString("Charlie"), 300),
+                            GenericRow.of(4, BinaryString.fromString(nullLiteral), 400));
+
             List<InternalRow> result =
                     writeThenRead(
                             options,
@@ -484,7 +484,7 @@ public void testCsvNullLiteralWriteRead() throws IOException {
                             "test_null_literal_" + nullLiteral.hashCode());
 
             // Verify results
-            assertThat(result).hasSize(3);
+            assertThat(result).hasSize(4);
             assertThat(result.get(0).getInt(0)).isEqualTo(1);
             assertThat(result.get(0).getString(1).toString()).isEqualTo("Alice");
             assertThat(result.get(0).isNullAt(2)).isTrue();
@@ -494,6 +494,9 @@ public void testCsvNullLiteralWriteRead() throws IOException {
             assertThat(result.get(2).getInt(0)).isEqualTo(3);
             assertThat(result.get(2).getString(1).toString()).isEqualTo("Charlie");
             assertThat(result.get(2).getInt(2)).isEqualTo(300);
+            assertThat(result.get(3).getInt(0)).isEqualTo(4);
+            assertThat(result.get(3).getString(1).toString()).isEqualTo(nullLiteral);
+            assertThat(result.get(3).getInt(2)).isEqualTo(400);
         }
     }