This is an automated email from the ASF dual-hosted git repository. raminqaf pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 33a8167b49a17c6f45e0e3f864c9de859aa36430 Author: Ramin Gharib <[email protected]> AuthorDate: Tue Nov 3 11:44:38 2026 +0100 [FLINK-41305][table-runtime] Support the raw line delimiter only for a string column and split UTF-8 on its bytes With 'raw.line-delimiter', the raw format split a message of any column type. It decoded the whole message with 'raw.charset', split the text and encoded each part again before converting it. Decoding turned invalid UTF-8 into U+FFFD, so a STRING column lost bytes that it keeps without a delimiter, and a VARIANT column accepted bytes that PARSE_JSON rejects. Splitting the bytes of a number or a BYTES value is not meaningful either, because those bytes can contain the delimiter. 'raw.line-delimiter' is now supported only for a CHAR, VARCHAR or STRING column. Any other column type fails with a ValidationException when a query reads or writes the table. Newline-delimited JSON can be read through a STRING column and PARSE_JSON. If 'raw.charset' is UTF-8, the message is split on the UTF-8 bytes of the delimiter and never decoded. In UTF-8 a delimiter's bytes can only match at a character boundary, so valid UTF-8 splits into the same parts as before. Other charsets keep the decoded split, because their delimiter bytes can match inside a character, as '|' does in GBK. An empty delimiter keeps it too, because it splits between characters. Generated-by: Claude Code 2.1.316 (Claude Opus 5.5) --- .../docs/connectors/table/formats/raw.md | 11 +- docs/content/docs/connectors/table/formats/raw.md | 12 +- .../raw/RawFormatDeserializationSchema.java | 128 ++++++++++++++++----- .../apache/flink/formats/raw/RawFormatFactory.java | 19 +++ .../table/formats/raw/RawFormatFactoryTest.java | 26 +++++ .../formats/raw/RawFormatLineDelimiterTest.java | 91 ++++++++------- 6 files changed, 204 insertions(+), 83 deletions(-) diff --git a/docs/content.zh/docs/connectors/table/formats/raw.md b/docs/content.zh/docs/connectors/table/formats/raw.md index 7bd20aba466..1d728c3fadb 100644 --- a/docs/content.zh/docs/connectors/table/formats/raw.md +++ b/docs/content.zh/docs/connectors/table/formats/raw.md @@ -110,8 +110,11 @@ Format 参数 <td>可选</td> <td style="word-wrap: break-word;">(无)</td> <td>String</td> - <td>指定行分隔符，用于在反序列化时将一条消息拆分为多行。设置后，每条消息将使用 - 'raw.charset' 解码，并按此分隔符切分，每段输出一条数据行。在序列化时，分隔符字节会被 + <td>指定行分隔符，用于在反序列化时将一条消息拆分为多行。Only supported for a <code>STRING</code>, + <code>VARCHAR</code> or <code>CHAR</code> column. When set, each incoming message is split by + this delimiter; one row is emitted per segment. If 'raw.charset' is UTF-8, the message is split on + the UTF-8 bytes of the delimiter without being decoded, so each segment keeps its original bytes. + Otherwise the message is decoded using 'raw.charset' before it is split. 在序列化时，分隔符字节会被 追加到每条序列化值的末尾。常用值为 '\n'（换行符）或 '||'。 <br><strong>注意：</strong>当序列化与反序列化使用相同的分隔符配置时，两者具有 round-trip 兼容性：序列化器追加的末尾分隔符会在反序列化时自动被去除。</td> @@ -191,6 +194,6 @@ Format 参数 </tbody> </table> -Note: combining `VARIANT` with `raw.line-delimiter` gives you newline-delimited JSON, where each line of a message -becomes one row. Without the delimiter, only the first JSON document of a message is read. +Note: a `VARIANT` column holds one JSON document per message. If a message holds more, only the first is read. To read +newline-delimited JSON, declare the column as `STRING` with `raw.line-delimiter` and convert each row with `PARSE_JSON`. diff --git a/docs/content/docs/connectors/table/formats/raw.md b/docs/content/docs/connectors/table/formats/raw.md index 0886317085a..2f183e53f44 100644 --- a/docs/content/docs/connectors/table/formats/raw.md +++ b/docs/content/docs/connectors/table/formats/raw.md @@ -111,8 +111,12 @@ Format Options <td style="word-wrap: break-word;">(none)</td> <td>String</td> <td>Specify the line delimiter for splitting incoming messages into multiple rows during - deserialization. When set, each incoming message is decoded using 'raw.charset' and then split - by this delimiter; one row is emitted per segment. During serialization, the delimiter bytes + deserialization. Only supported for a <code>STRING</code>, <code>VARCHAR</code> or <code>CHAR</code> + column. When set, each incoming message is split by this delimiter; one row is emitted + per segment. If 'raw.charset' is UTF-8, the message is split on the UTF-8 bytes of the delimiter + without being decoded, so each segment keeps its original bytes. Otherwise the message is decoded + using 'raw.charset' before it is split. + During serialization, the delimiter bytes are appended to each serialized value. Common values are '\n' (newline) or '||'. <br><strong>Note:</strong> When the same delimiter is configured for both serialization and deserialization, the two are round-trip compatible: a trailing delimiter appended by the @@ -193,6 +197,6 @@ The table below details the SQL types the format supports, including details of </tbody> </table> -Note: combining `VARIANT` with `raw.line-delimiter` gives you newline-delimited JSON, where each line of a message -becomes one row. Without the delimiter, only the first JSON document of a message is read. +Note: a `VARIANT` column holds one JSON document per message. If a message holds more, only the first is read. To read +newline-delimited JSON, declare the column as `STRING` with `raw.line-delimiter` and convert each row with `PARSE_JSON`. diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java index e6e894b5c61..b89881eee49 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java @@ -39,7 +39,9 @@ import java.io.IOException; import java.io.Serializable; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.Objects; +import java.util.function.Consumer; import java.util.regex.Pattern; import static org.apache.flink.util.Preconditions.checkNotNull; @@ -63,12 +65,6 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row @Nullable private final String lineDelimiter; - /** - * Pre-compiled pattern for splitting by {@link #lineDelimiter}, or {@code null} if no - * delimiter. Note: this field and {@link #lineDelimiter} are either both null or both non-null. - */ - @Nullable private final Pattern lineDelimiterPattern; - private final DeserializationRuntimeConverter converter; private final DataLengthValidator validator; @@ -89,13 +85,11 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row @Nullable String lineDelimiter) { this.deserializedType = checkNotNull(deserializedType); this.producedTypeInfo = checkNotNull(producedTypeInfo); - this.converter = createConverter(deserializedType, charsetName, isBigEndian); + this.converter = createConverter(deserializedType, charsetName, isBigEndian, lineDelimiter); this.validator = createDataLengthValidator(deserializedType); this.charsetName = charsetName; this.isBigEndian = isBigEndian; this.lineDelimiter = lineDelimiter; - this.lineDelimiterPattern = - lineDelimiter != null ? Pattern.compile(Pattern.quote(lineDelimiter)) : null; } @Override @@ -133,24 +127,13 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row return; } - Charset charset = Charset.forName(charsetName); - String decoded = new String(message, charset); - // Use pre-compiled pattern. Split with -1 to keep intentional empty middle segments, - // but strip the single trailing empty string produced when the message ends with the - // delimiter (e.g. a serializer that appends one delimiter per row). - String[] parts = lineDelimiterPattern.split(decoded, -1); - int count = parts.length; - if (count > 0 && parts[count - 1].isEmpty()) { - count--; - } - for (int i = 0; i < count; i++) { - byte[] partBytes = parts[i].getBytes(charset); - validator.validate(partBytes); - Object field = converter.convert(partBytes); - GenericRowData rowData = new GenericRowData(1); - rowData.setField(0, field); - out.collect(rowData); - } + converter.convertLines( + message, + field -> { + final GenericRowData rowData = new GenericRowData(1); + rowData.setField(0, field); + out.collect(rowData); + }); } @Override @@ -194,16 +177,29 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row default void open() {} Object convert(byte[] data) throws IOException; + + /** + * Converts a message that holds one value per line. Only a string column takes a line + * delimiter, so only the string converter splits the message. + */ + default void convertLines(byte[] message, Consumer<Object> out) throws IOException { + out.accept(convert(message)); + } } /** Creates a runtime converter. */ private static DeserializationRuntimeConverter createConverter( - LogicalType type, String charsetName, boolean isBigEndian) { + LogicalType type, + String charsetName, + boolean isBigEndian, + @Nullable String lineDelimiter) { switch (type.getTypeRoot()) { case CHAR: case VARCHAR: - return createStringConverter(charsetName); + return lineDelimiter == null + ? createStringConverter(charsetName) + : createDelimitedStringConverter(charsetName, lineDelimiter); case VARBINARY: case BINARY: @@ -283,6 +279,80 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row }; } + /** + * Creates a converter for a string column with a line delimiter, which turns a message into one + * string per line. In UTF-8 the bytes of a delimiter can only match at a character boundary, so + * the message is split without decoding it. In other charsets they can match inside a + * character, and an empty delimiter splits between characters, so both decode the message + * first. + */ + private static DeserializationRuntimeConverter createDelimitedStringConverter( + final String charsetName, final String lineDelimiter) { + final DeserializationRuntimeConverter stringConverter = createStringConverter(charsetName); + final boolean splitBytes = + !lineDelimiter.isEmpty() + && StandardCharsets.UTF_8.equals(Charset.forName(charsetName)); + final byte[] delimiter = lineDelimiter.getBytes(StandardCharsets.UTF_8); + final Pattern pattern = Pattern.compile(Pattern.quote(lineDelimiter)); + return new DeserializationRuntimeConverter() { + private static final long serialVersionUID = 1L; + private transient Charset charset; + + @Override + public void open() { + stringConverter.open(); + charset = Charset.forName(charsetName); + } + + @Override + public Object convert(byte[] data) throws IOException { + return stringConverter.convert(data); + } + + @Override + public void convertLines(byte[] message, Consumer<Object> out) throws IOException { + if (splitBytes) { + int start = 0; + int end; + while ((end = indexOf(message, start)) >= 0) { + out.accept(convert(Arrays.copyOfRange(message, start, end))); + start = end + delimiter.length; + } + // Like the decoded split below, drop the empty part after a trailing delimiter. + if (start < message.length) { + out.accept(convert(Arrays.copyOfRange(message, start, message.length))); + } + return; + } + // Split with -1 to keep intentional empty middle segments, but strip the single + // trailing empty string produced when the message ends with the delimiter (e.g. a + // serializer that appends one delimiter per row). + final String[] parts = pattern.split(new String(message, charset), -1); + int count = parts.length; + if (count > 0 && parts[count - 1].isEmpty()) { + count--; + } + for (int i = 0; i < count; i++) { + out.accept(convert(parts[i].getBytes(charset))); + } + } + + /** Returns the position of the next delimiter at or after {@code from}, or -1. */ + private int indexOf(byte[] message, int from) { + for (int i = from; i <= message.length - delimiter.length; i++) { + int j = 0; + while (j < delimiter.length && message[i + j] == delimiter[j]) { + j++; + } + if (j == delimiter.length) { + return i; + } + } + return -1; + } + }; + } + /** * Creates a converter that parses UTF-8 JSON bytes into a {@link Variant} like {@code * PARSE_JSON}, so invalid UTF-8 and duplicate object keys are rejected. diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java index 59fd275ba6c..bf2899d64ec 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java @@ -37,6 +37,7 @@ import org.apache.flink.table.factories.FactoryUtil; import org.apache.flink.table.factories.SerializationFormatFactory; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeFamily; import org.apache.flink.table.types.logical.LogicalTypeRoot; import org.apache.flink.table.types.logical.RowType; @@ -93,6 +94,7 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa DynamicTableSource.Context context, DataType producedDataType) { final RowType physicalRowType = (RowType) producedDataType.getLogicalType(); final LogicalType fieldType = validateAndExtractSingleField(physicalRowType); + checkLineDelimiter(fieldType, lineDelimiter); checkVariantCharset(fieldType, charsetName); final TypeInformation<RowData> producedTypeInfo = context.createTypeInformation(producedDataType); @@ -126,6 +128,7 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa DynamicTableSink.Context context, DataType consumedDataType) { final RowType physicalRowType = (RowType) consumedDataType.getLogicalType(); final LogicalType fieldType = validateAndExtractSingleField(physicalRowType); + checkLineDelimiter(fieldType, lineDelimiter); checkVariantCharset(fieldType, charsetName); return new RawFormatSerializationSchema( fieldType, charsetName, isBigEndian, lineDelimiter.orElse(null)); @@ -166,6 +169,22 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa } } + /** + * Checks that a line delimiter is only set for a string column. The bytes of other types can + * contain the delimiter, so splitting them would cut a value apart. + */ + private static void checkLineDelimiter(LogicalType fieldType, Optional<String> lineDelimiter) { + if (lineDelimiter.isPresent() && !fieldType.is(LogicalTypeFamily.CHARACTER_STRING)) { + throw new ValidationException( + String.format( + "The 'raw' format supports '%s.%s' only for a string column, " + + "but the column type is '%s'.", + IDENTIFIER, + RawFormatOptions.LINE_DELIMITER.key(), + fieldType.asSummaryString())); + } + } + /** * Checks that a VARIANT column uses UTF-8, which RFC 8259 requires for JSON exchanged between * systems and which {@code PARSE_JSON} reads. diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java index 11bf1b74d52..92e95efa688 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java @@ -33,13 +33,17 @@ import org.apache.flink.table.factories.TestDynamicTableFactory; import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext; import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext; import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.RowType; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; import java.util.HashMap; import java.util.Map; import java.util.function.Consumer; +import java.util.stream.Stream; import static java.nio.charset.StandardCharsets.UTF_16; import static java.nio.charset.StandardCharsets.UTF_8; @@ -217,6 +221,28 @@ class RawFormatFactoryTest { .hasMessage(expectedError); } + @ParameterizedTest + @MethodSource("nonStringTypes") + void testLineDelimiterRequiresStringColumn(final DataType type) { + final ResolvedSchema schema = ResolvedSchema.of(Column.physical("field1", type)); + final Map<String, String> tableOptions = + getModifiedOptions(options -> options.put("raw.line-delimiter", "\n")); + final String expectedError = + String.format( + "The 'raw' format supports 'raw.line-delimiter' only for a string column, " + + "but the column type is '%s'.", + type.getLogicalType().asSummaryString()); + + assertThatThrownBy(() -> createDeserializationSchema(schema, tableOptions)) + .hasMessage(expectedError); + assertThatThrownBy(() -> createSerializationSchema(schema, tableOptions)) + .hasMessage(expectedError); + } + + static Stream<DataType> nonStringTypes() { + return Stream.of(DataTypes.VARIANT(), DataTypes.BYTES(), DataTypes.INT()); + } + @Test void testLineDelimiterOption() { final Map<String, String> tableOptions = diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java index e12a28f958d..0c138ab33e1 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java @@ -30,19 +30,17 @@ import org.apache.flink.table.data.StringData; import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.table.types.logical.VariantType; import org.apache.flink.types.DeserializationException; -import org.apache.flink.types.variant.BinaryVariantInternalBuilder; import org.apache.flink.util.Collector; +import org.apache.flink.util.InstantiationUtil; import org.apache.flink.util.UserCodeClassLoader; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParseException; import org.junit.jupiter.api.Test; -import java.io.IOException; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -80,13 +78,15 @@ class RawFormatLineDelimiterTest { @Test void testDeserializeWithNewlineDelimiter_multipleRows() throws Exception { + // Cloned as Flink ships the schema, which also checks that the UTF-8 splitter serializes. RawFormatDeserializationSchema schema = - new RawFormatDeserializationSchema( - STRING_TYPE, - TypeInformation.of(RowData.class), - StandardCharsets.UTF_8.name(), - true, - "\n"); + InstantiationUtil.clone( + new RawFormatDeserializationSchema( + STRING_TYPE, + TypeInformation.of(RowData.class), + StandardCharsets.UTF_8.name(), + true, + "\n")); openDeser(schema); byte[] message = "line1\nline2\nline3".getBytes(StandardCharsets.UTF_8); @@ -134,20 +134,41 @@ class RawFormatLineDelimiterTest { @Test void testDeserializeWithGbkCharset() throws Exception { Charset gbk = Charset.forName("GBK"); - String original = "你好\n世界"; + // The GBK encoding of 亅 is 0x81 0x7C, so its second byte equals the delimiter '|'. + String original = "你好亅|世界"; byte[] message = original.getBytes(gbk); + // Cloned as Flink ships the schema, which also checks that the decoding splitter + // serializes. RawFormatDeserializationSchema schema = - new RawFormatDeserializationSchema( - STRING_TYPE, TypeInformation.of(RowData.class), "GBK", true, "\n"); + InstantiationUtil.clone( + new RawFormatDeserializationSchema( + STRING_TYPE, TypeInformation.of(RowData.class), "GBK", true, "|")); openDeser(schema); List<RowData> rows = collectRows(schema, message); assertThat(rows).hasSize(2); - assertThat(rows.get(0).getString(0)).hasToString("你好"); + assertThat(rows.get(0).getString(0)).hasToString("你好亅"); assertThat(rows.get(1).getString(0)).hasToString("世界"); } + @Test + void testDeserializeWithDelimiterKeepsInvalidUtf8() throws Exception { + RawFormatDeserializationSchema schema = + new RawFormatDeserializationSchema( + STRING_TYPE, + TypeInformation.of(RowData.class), + StandardCharsets.UTF_8.name(), + true, + "\n"); + openDeser(schema); + + byte[] message = {'a', (byte) 0xFF, '\n', (byte) 0x80}; + assertThat(collectRows(schema, message)) + .extracting(row -> row.getString(0).toBytes()) + .containsExactly(new byte[] {'a', (byte) 0xFF}, new byte[] {(byte) 0x80}); + } + // ----------------------------------------------------------------------- // Serialization tests // ----------------------------------------------------------------------- @@ -250,32 +271,22 @@ class RawFormatLineDelimiterTest { } @Test - void testRoundTripNewlineDelimitedJsonAsVariant() throws Exception { - RawFormatSerializationSchema ser = - new RawFormatSerializationSchema( - VARIANT_TYPE, StandardCharsets.UTF_8.name(), true, "\n"); - openSer(ser); - - RawFormatDeserializationSchema deser = + void testDeserializeWithMultiByteDelimiter() throws Exception { + RawFormatDeserializationSchema schema = new RawFormatDeserializationSchema( - VARIANT_TYPE, + STRING_TYPE, TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "\n"); - openDeser(deser); - - byte[] stream = - concat( - ser.serialize(buildVariantRow("{\"id\":1}")), - ser.serialize(buildVariantRow("{\"id\":2}"))); - assertThat(new String(stream, StandardCharsets.UTF_8)) - .isEqualTo("{\"id\":1}\n{\"id\":2}\n"); + "é"); + openDeser(schema); - List<RowData> rows = collectRows(deser, stream); - assertThat(rows).hasSize(2); - assertThat(rows.get(0).getVariant(0).getField("id").getByte()).isEqualTo((byte) 1); - assertThat(rows.get(1).getVariant(0).getField("id").getByte()).isEqualTo((byte) 2); + // The delimiter é is c3 a9 and the data holds ü, which is c3 bc, so a match on the first + // byte alone is not enough. + byte[] message = "aüébüé".getBytes(StandardCharsets.UTF_8); + assertThat(collectRows(schema, message)) + .extracting(row -> row.getString(0).toString()) + .containsExactly("aü", "bü"); } @Test @@ -307,12 +318,6 @@ class RawFormatLineDelimiterTest { // Helpers // ----------------------------------------------------------------------- - private static byte[] concat(byte[] first, byte[] second) { - byte[] result = Arrays.copyOf(first, first.length + second.length); - System.arraycopy(second, 0, result, first.length, second.length); - return result; - } - private void openDeser(RawFormatDeserializationSchema schema) throws Exception { schema.open( new DeserializationSchema.InitializationContext() { @@ -365,10 +370,4 @@ class RawFormatLineDelimiterTest { row.setField(0, StringData.fromString(value)); return row; } - - private RowData buildVariantRow(String json) throws IOException { - GenericRowData row = new GenericRowData(1); - row.setField(0, BinaryVariantInternalBuilder.parseJson(json, false)); - return row; - } }
