[format] Escape the escape character when writing CSV

paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFormatWriter.java

@@ -98,22 +98,26 @@ private String escapeField(String field) {
             return csvOptions.nullLiteral();
         }
 
+        String quote = csvOptions.quoteCharacter();
+        String escape = csvOptions.escapeCharacter();
+        boolean escapable = !escape.isEmpty();
+
         // Optimized escaping with early exit checks
         boolean needsQuoting =
                 field.indexOf(csvOptions.fieldDelimiter().charAt(0)) >= 0
                         || field.indexOf(csvOptions.lineDelimiter().charAt(0)) >= 0
-                        || field.indexOf(csvOptions.quoteCharacter().charAt(0)) >= 0;
+                        || field.indexOf(quote.charAt(0)) >= 0
+                        || (escapable && field.indexOf(escape.charAt(0)) >= 0);
 
         if (!needsQuoting) {
             return field;
         }
 
-        // Only escape if needed
-        String escaped =
-                field.replace(
-                        csvOptions.quoteCharacter(),
-                        csvOptions.escapeCharacter() + csvOptions.quoteCharacter());
-        return csvOptions.quoteCharacter() + escaped + csvOptions.quoteCharacter();
+        // Only escape if needed. The escape character goes first: CsvParser drops an escape
+        // character that is not followed by a quote or another escape, and escaping the quotes
+        // first would double the escape characters inserted for them.
+        String escaped = escapable ? field.replace(escape, escape + escape) : field;
+        return quote + escaped.replace(quote, escape + quote) + quote;
     }
 
     /** Optimized string casting with caching and fast paths for common types. */

paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java

@@ -371,9 +371,11 @@ public void testCsvEscapeCharacterWriteRead() throws IOException {
             // Verify results
             assertThat(result).hasSize(3);
             assertThat(result.get(0).getInt(0)).isEqualTo(1);
+            assertThat(result.get(0).getString(1).toString()).isEqualTo("Value\"With\"Quotes");
             assertThat(result.get(1).getInt(0)).isEqualTo(2);
             assertThat(result.get(1).getString(1).toString()).isEqualTo("Normal Value");
             assertThat(result.get(2).getInt(0)).isEqualTo(3);
+            assertThat(result.get(2).getString(1).toString()).isEqualTo("Special\\Characters");
         }
     }
 
@@ -826,6 +828,31 @@ private String[] parse(String csvLine) throws IOException {
      * Performs a complete write-read test with the given options and test data. Returns the data
      * that was read back for further verification.
      */
+    @Test
+    public void testFieldsContainingTheEscapeCharacterRoundTrip() throws IOException {
+        RowType rowType = DataTypes.ROW(DataTypes.INT().notNull(), DataTypes.STRING());
+        // every one of these is written unquoted or half-quoted unless the escape character is
+        // itself escaped, and CsvParser then drops it
+        String[] inputs = {
+            "Special\\Characters", "trailing\\", "a,b\\", "\\\\double", "\\\"quoteAfterEscape"
+        };
+
+        List<InternalRow> testData = new ArrayList<>();
+        for (int i = 0; i < inputs.length; i++) {
+            testData.add(GenericRow.of(i, BinaryString.fromString(inputs[i])));
+        }
+
+        List<InternalRow> result =
+                writeThenRead(new Options(), rowType, rowType, testData, "escape_round_trip");
+
+        assertThat(result).hasSize(inputs.length);
+        for (int i = 0; i < inputs.length; i++) {
+            assertThat(result.get(i).getInt(0)).isEqualTo(i);
+            assertThat(result.get(i).getString(1)).isNotNull();
+            assertThat(result.get(i).getString(1).toString()).isEqualTo(inputs[i]);
+        }
+    }
+
     private List<InternalRow> writeThenRead(
             Options options,
             RowType fullRowType,