[format] Reject multi-character csv delimiter, quote and escape

docs/docs/concepts/spec/fileformat.md

@@ -151,8 +151,8 @@ Format Options:
 | --- | --- | --- | --- |
 | `csv.field-delimiter` | `,` | String | Field delimiter character (`','` by default), must be single character. You can use backslash to specify special characters, e.g. `'\t'` represents the tab character. |
 | `csv.line-delimiter` | `\n` | String | The line delimiter for CSV format |
-| `csv.quote-character` | `"` | String | Quote character for enclosing field values (`"` by default). |
-| `csv.escape-character` | `\` | String | The escape character for CSV format. |
+| `csv.quote-character` | `"` | String | Quote character for enclosing field values (`"` by default), must be single character. |
+| `csv.escape-character` | `\` | String | The escape character for CSV format, must be single character. |
 | `csv.include-header` | false | Boolean | Whether to include header in CSV files. |
 | `csv.null-literal` | `""` | String | Null literal string that is interpreted as a null value (disabled by default). |
 | `csv.mode` | `PERMISSIVE` | String | Allows a mode for dealing with corrupt records during reading. Currently supported values are `'PERMISSIVE'`, `'DROPMALFORMED'` and `'FAILFAST'`: <ul> <li>Option `'PERMISSIVE'` sets malformed fields to null.</li> <li>Option `'DROPMALFORMED'` ignores the whole corrupted records.</li> <li>Option `'FAILFAST'` throws an exception when it meets corrupted records.</li> </ul> |

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

@@ -23,6 +23,7 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.options.description.DescribedEnum;
 import org.apache.paimon.options.description.InlineElement;
+import org.apache.paimon.utils.Preconditions;
 
 import static org.apache.paimon.options.description.TextElement.text;
 
@@ -88,15 +89,31 @@ public class CsvOptions {
     private final Mode mode;
 
     public CsvOptions(Options options) {
-        this.fieldDelimiter = options.get(FIELD_DELIMITER);
+        this.fieldDelimiter = singleCharacter(options, FIELD_DELIMITER);
         this.lineDelimiter = options.get(LINE_DELIMITER);
         this.nullLiteral = options.get(NULL_LITERAL);
         this.includeHeader = options.get(INCLUDE_HEADER);
-        this.quoteCharacter = options.get(QUOTE_CHARACTER);
-        this.escapeCharacter = options.get(ESCAPE_CHARACTER);
+        this.quoteCharacter = singleCharacter(options, QUOTE_CHARACTER);
+        this.escapeCharacter = singleCharacter(options, ESCAPE_CHARACTER);
         this.mode = options.get(MODE);
     }
 
+    /**
+     * {@link CsvParser} keeps only the first character of these options while the writer emits the
+     * whole string, so a longer value is written and read differently. The line delimiter is
+     * deliberately not restricted here: {@code CustomLineReader} matches all of its bytes for the
+     * Paimon implementation.
+     */
+    private static String singleCharacter(Options options, ConfigOption<String> option) {
+        String value = options.get(option);
+        Preconditions.checkArgument(
+                value.length() == 1,
+                "'%s' must be a single character, but was '%s'.",
+                option.key(),
+                value);
+        return value;
+    }
+
     public String fieldDelimiter() {
         return fieldDelimiter;
     }

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

@@ -34,6 +34,7 @@
 import org.apache.paimon.format.HadoopCompressionType;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.options.ConfigOption;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.types.DataTypes;
@@ -54,6 +55,7 @@
 
 import static org.apache.paimon.data.BinaryString.fromString;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link CsvFileFormat}. */
@@ -907,6 +909,30 @@ private List<InternalRow> writeThenRead(
         return read(format, fullRowType, rowType, testFile);
     }
 
+    @Test
+    public void testSingleCharacterOptionsAreEnforced() {
+        // The writer emits the whole option string while CsvParser keeps only charAt(0), so a
+        // multi-character value silently wrote one delimiter and read back another.
+        for (ConfigOption<String> option :
+                Arrays.asList(
+                        CsvOptions.FIELD_DELIMITER,
+                        CsvOptions.QUOTE_CHARACTER,
+                        CsvOptions.ESCAPE_CHARACTER)) {
+            for (String bad : Arrays.asList("ab", "")) {
+                Options options = new Options();
+                options.set(option, bad);
+                assertThatThrownBy(() -> new CsvOptions(options))
+                        .isInstanceOf(IllegalArgumentException.class)
+                        .hasMessageContaining(option.key());
+            }
+        }
+
+        // A multi-character line delimiter stays supported; CustomLineReader matches all of it.
+        Options multiCharLine = new Options();
+        multiCharLine.set(CsvOptions.LINE_DELIMITER, "|||");
+        assertThatCode(() -> new CsvOptions(multiCharLine)).doesNotThrowAnyException();
+    }
+
     /** Writes the given data to a new CSV file and returns its path. */
     private Path write(
             FileFormat format, RowType rowType, List<InternalRow> testData, String testPrefix)