This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a3907798ea [flink] Fix CSV delimiter fallback for format tables (#9178)
a3907798ea is described below
commit a3907798ea0aeb2d7813e47dd5ed74c21878f333
Author: Arnav Balyan <[email protected]>
AuthorDate: Wed Aug 12 12:55:16 2026 +0530
[flink] Fix CSV delimiter fallback for format tables (#9178)
---
.../apache/paimon/flink/FormatCatalogTable.java | 11 +++++-
.../paimon/flink/FormatCatalogTableTest.java | 46 ++++++++++++++++++++++
2 files changed, 55 insertions(+), 2 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FormatCatalogTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FormatCatalogTable.java
index c7bf660676..5dbce5f0a4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FormatCatalogTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FormatCatalogTable.java
@@ -18,6 +18,8 @@
package org.apache.paimon.flink;
+import org.apache.paimon.format.csv.CsvOptions;
+import org.apache.paimon.options.Options;
import org.apache.paimon.table.FormatTable;
import org.apache.flink.table.api.Schema;
@@ -94,8 +96,13 @@ public class FormatCatalogTable implements CatalogTable {
cachedOptions.put(k, v);
}
});
- if (options.containsKey("field-delimiter")) {
- cachedOptions.put("csv.field-delimiter",
options.get("field-delimiter"));
+ if ("csv".equals(format)) {
+ Options csvOptions = Options.fromMap(options);
+ if (csvOptions.contains(CsvOptions.FIELD_DELIMITER)) {
+ cachedOptions.put(
+ CsvOptions.FIELD_DELIMITER.key(),
+ csvOptions.get(CsvOptions.FIELD_DELIMITER));
+ }
}
cachedOptions.put(CONNECTOR.key(), "filesystem");
cachedOptions.put(PATH.key(), table.location());
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FormatCatalogTableTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FormatCatalogTableTest.java
new file mode 100644
index 0000000000..4c3d5711fb
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FormatCatalogTableTest.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.format.csv.CsvOptions;
+import org.apache.paimon.table.FormatTable;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class FormatCatalogTableTest {
+
+ @ParameterizedTest
+ @ValueSource(strings = {"field-delimiter", "seq", "delimiter"})
+ void testCsvFieldDelimiterFallbackKeys(String fallbackKey) {
+ FormatTable table = mock(FormatTable.class);
+ when(table.format()).thenReturn(FormatTable.Format.CSV);
+ when(table.options()).thenReturn(Collections.singletonMap(fallbackKey,
";"));
+ when(table.location()).thenReturn("file:/tmp/t");
+
+ assertThat(new FormatCatalogTable(table).getOptions())
+ .containsEntry(CsvOptions.FIELD_DELIMITER.key(), ";");
+ }
+}
