[format] Reject csv values containing a row separator

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.casting.CastExecutors;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.format.text.AbstractTextFileWriter;
+import org.apache.paimon.format.text.TextLineReader;
 import org.apache.paimon.fs.PositionOutputStream;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypeRoot;
@@ -43,13 +44,19 @@ public class CsvFormatWriter extends AbstractTextFileWriter {
     private final CsvOptions csvOptions;
     private boolean headerWritten = false;
     private final StringBuilder stringBuilder;
+    private final String[] fieldNames;
+    // CR and LF only split a line when StandardLineReader is in use, which TextLineReader picks
+    // solely from the delimiter; under a custom delimiter they are ordinary bytes.
+    private final boolean lineBreakSplitsLine;
 
     public CsvFormatWriter(
             PositionOutputStream out, RowType rowType, CsvOptions options, String compression)
             throws IOException {
         super(out, rowType, compression);
         this.csvOptions = options;
         this.stringBuilder = new StringBuilder();
+        this.fieldNames = rowType.getFieldNames().toArray(new String[0]);
+        this.lineBreakSplitsLine = TextLineReader.isDefaultDelimiter(options.lineDelimiter());
     }
 
     @Override
@@ -71,7 +78,8 @@ public void addElement(InternalRow element) throws IOException {
 
             Object value =
                     InternalRow.createFieldGetter(rowType.getTypeAt(i), i).getFieldOrNull(element);
-            String fieldValue = escapeField(castToStringOptimized(value, rowType.getTypeAt(i)));
+            String fieldValue =
+                    escapeField(castToStringOptimized(value, rowType.getTypeAt(i)), fieldNames[i]);
             stringBuilder.append(fieldValue);
         }
         stringBuilder.append(csvOptions.lineDelimiter());
@@ -87,28 +95,46 @@ private void writeHeader() throws IOException {
             if (i > 0) {
                 stringBuilder.append(csvOptions.fieldDelimiter());
             }
-            stringBuilder.append(escapeField(rowType.getFieldNames().get(i)));
+            stringBuilder.append(escapeField(fieldNames[i], fieldNames[i]));
         }
         stringBuilder.append(csvOptions.lineDelimiter());
         writer.write(stringBuilder.toString());
     }
 
-    private String escapeField(String field) {
+    private String escapeField(String field, String fieldName) {
         if (field == null) {
             return csvOptions.nullLiteral();
         }
 
         String quote = csvOptions.quoteCharacter();
         String escape = csvOptions.escapeCharacter();
-        boolean escapable = !escape.isEmpty();
+        String lineDelimiter = csvOptions.lineDelimiter();
+
+        // A value carrying the row separator cannot be read back. The line readers match it
+        // without tracking quotes, and a split boundary may fall inside the value, so quoting
+        // cannot rescue it without giving up splittability. Refuse the value rather than write a
+        // file that reads back as extra rows. CR and LF only count when they are the separator:
+        // under a custom delimiter CustomLineReader treats them as ordinary bytes, which is the
+        // documented way to carry a line break inside a value.
+        if (field.contains(lineDelimiter)
+                || (lineBreakSplitsLine
+                        && (field.indexOf('\r') >= 0 || field.indexOf('\n') >= 0))) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Column '%s' contains the row separator, which the CSV format cannot "
+                                    + "represent: '%s'",
+                            fieldName, truncate(field)));
+        }
 
-        // Optimized escaping with early exit checks
+        // Optimized escaping with early exit checks. A value that merely starts a delimiter match
+        // still has to be quoted: CustomLineReader is leftmost-match, so the delimiter appended
+        // after the row would complete a match begun by the value's own trailing bytes.
         boolean needsQuoting =
                 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);
+                        || field.contains(csvOptions.fieldDelimiter())
+                        || field.indexOf(lineDelimiter.charAt(0)) >= 0
+                        || field.contains(quote)
+                        || field.contains(escape);
 
         if (!needsQuoting) {
             return field;
@@ -117,10 +143,15 @@ private String escapeField(String field) {
         // 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;
+        String escaped = field.replace(escape, escape + escape);
         return quote + escaped.replace(quote, escape + quote) + quote;
     }
 
+    /** Keeps an unbounded STRING value from turning into an unbounded exception message. */
+    private static String truncate(String field) {
+        return field.length() <= 64 ? field : field.substring(0, 64) + "...";
+    }
+
     /** Optimized string casting with caching and fast paths for common types. */
     private String castToStringOptimized(Object value, DataType dataType) {
         if (value == null) {

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

@@ -91,6 +91,8 @@ public class CsvOptions {
     public CsvOptions(Options options) {
         this.fieldDelimiter = singleCharacter(options, FIELD_DELIMITER);
         this.lineDelimiter = options.get(LINE_DELIMITER);
+        Preconditions.checkArgument(
+                !lineDelimiter.isEmpty(), "'%s' must not be empty.", LINE_DELIMITER.key());
         this.nullLiteral = options.get(NULL_LITERAL);
         this.includeHeader = options.get(INCLUDE_HEADER);
         this.quoteCharacter = singleCharacter(options, QUOTE_CHARACTER);

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

@@ -49,6 +49,7 @@
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ThreadLocalRandom;
@@ -896,6 +897,52 @@ public void testFieldsContainingTheEscapeCharacterRoundTrip() throws IOException
         }
     }
 
+    @Test
+    public void testValueContainingRowSeparatorIsRejected() throws IOException {
+        // Quoting cannot rescue an embedded row separator: the line readers split on it without
+        // tracking quotes, and a split boundary may fall inside the value, so the row used to come
+        // back as two rows with NULLs and COUNT(*) changed.
+        RowType rowType = DataTypes.ROW(DataTypes.INT().notNull(), DataTypes.STRING());
+        for (String value : Arrays.asList("hello\nworld", "hello\rworld")) {
+            List<InternalRow> row =
+                    Collections.singletonList(GenericRow.of(1, BinaryString.fromString(value)));
+            assertThatThrownBy(
+                            () ->
+                                    writeThenRead(
+                                            new Options(), rowType, rowType, row, "row_separator"))
+                    .isInstanceOf(IllegalArgumentException.class)
+                    .hasMessageContaining("f1");
+        }
+
+        // The configured line delimiter is a separator too, even when it is not CR or LF.
+        Options customLine = new Options();
+        customLine.set(CsvOptions.LINE_DELIMITER, "|||");
+        List<InternalRow> pipes =
+                Collections.singletonList(GenericRow.of(1, BinaryString.fromString("a|||b")));
+        assertThatThrownBy(
+                        () -> writeThenRead(customLine, rowType, rowType, pipes, "row_separator"))
+                .isInstanceOf(IllegalArgumentException.class);
+
+        // A value that merely begins a delimiter match must still round-trip: CustomLineReader is
+        // leftmost-match, so the delimiter appended after the row would otherwise complete a match
+        // started by the value's own trailing bytes.
+        List<InternalRow> onePipe =
+                Collections.singletonList(GenericRow.of(1, BinaryString.fromString("x|")));
+        List<InternalRow> readBack =
+                writeThenRead(customLine, rowType, rowType, onePipe, "row_separator");
+        assertThat(readBack).hasSize(1);
+        assertThat(readBack.get(0).getString(1).toString()).isEqualTo("x|");
+
+        // Under a custom delimiter a line break is an ordinary byte, which is the documented way
+        // to carry one inside a value; it must not be rejected.
+        List<InternalRow> withBreak =
+                Collections.singletonList(GenericRow.of(1, BinaryString.fromString("a\nb")));
+        List<InternalRow> breakReadBack =
+                writeThenRead(customLine, rowType, rowType, withBreak, "row_separator");
+        assertThat(breakReadBack).hasSize(1);
+        assertThat(breakReadBack.get(0).getString(1).toString()).isEqualTo("a\nb");
+    }
+
     private List<InternalRow> writeThenRead(
             Options options,
             RowType fullRowType,