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 f6e3ca45cf3d..b9a7d37fffd5 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 @@ -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,6 +44,10 @@ 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) @@ -50,6 +55,8 @@ public CsvFormatWriter( 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) { diff --git a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvOptions.java b/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvOptions.java index e04f5cffc29a..8c923ac0f2f0 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvOptions.java +++ b/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); 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 ba2ebccb98b2..d2c26deb0f51 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 @@ -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 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 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 onePipe = + Collections.singletonList(GenericRow.of(1, BinaryString.fromString("x|"))); + List 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 withBreak = + Collections.singletonList(GenericRow.of(1, BinaryString.fromString("a\nb"))); + List breakReadBack = + writeThenRead(customLine, rowType, rowType, withBreak, "row_separator"); + assertThat(breakReadBack).hasSize(1); + assertThat(breakReadBack.get(0).getString(1).toString()).isEqualTo("a\nb"); + } + private List writeThenRead( Options options, RowType fullRowType,