[FLINK-39401] Extend raw format to support line-delimiter option

[featzhang]

Summary

This PR extends the raw format to support a new optional raw.line-delimiter config option.

When raw.line-delimiter is set:

• Deserialization: each incoming message is decoded to a string using raw.charset, split by the delimiter (String.split(Pattern.quote(delimiter), -1)), and one RowData is emitted per segment via deserialize(byte[], Collector<T>).
• Serialization: the delimiter bytes are appended to the serialized value.

When raw.line-delimiter is not set, all existing behavior is preserved exactly (backward compatible).

Example SQL

CREATE TABLE nginx_log (log STRING) WITH (
  'connector'          = 'kafka',
  'topic'              = 'nginx_log',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format'             = 'raw',
  'raw.line-delimiter' = '\n'
);
-- Each Kafka message "line1\nline2\nline3" produces 3 rows
SELECT log FROM nginx_log;

Changes

Class	Change
RawFormatOptions	Add LINE_DELIMITER ConfigOption<String> with no default value
RawFormatFactory	Read new option; pass to schema builders; register in optionalOptions()
RawFormatDeserializationSchema	Override deserialize(byte[], Collector) to split by delimiter when set; add lineDelimiter field to equals/hashCode
RawFormatSerializationSchema	Append delimiter bytes when set; add lineDelimiter field to equals/hashCode
RawFormatFactoryTest	Add testLineDelimiterOption()
RawFormatLineDelimiterTest	New test class: 9 tests covering splitting, GBK charset, null message, serialization appending, null row

Test Plan

• RawFormatLineDelimiterTest (9 tests):
  • No delimiter → single row (regression)
  • \n delimiter → 3 rows
  • Custom multi-char delimiter || → 3 rows
  • Null message with delimiter → 0 rows
  • GBK charset with \n delimiter → correct splitting
  • Serialization without delimiter → no change
  • Serialization with \n → appends \n
  • Serialization with || → appends ||
  • Serialization of null row → returns null
• RawFormatFactoryTest.testLineDelimiterOption(): verifies factory produces schemas with correct delimiter
• All 50 existing raw format tests continue to pass

[flinkbot]

CI report:

• 8205f0c Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[featzhang]

@flinkbot run azure

[rmetzger]

@featzhang Thanks for this contribution. The commit message in the PR should include the Jira ticket id.

[spuru9]

Thanks for the Jira note on split(-1) — makes sense for preserving intentional empty middle segments.
One thing I noticed though: the serializer appends the delimiter to every row, so a message it produces always ends with the delimiter. When that message is read back with the same config, split(-1) will emit a trailing empty row. So the two halves of the feature are subtly incompatible when used together on the same table.

When added a round-trip test:
- Serialize "hello" → produces "hello\n"
- Deserialize "hello\n" → expected 1 row, got 2

Would be good to note in the PR about the limitation.

[spuru9]

raw.line-delimiter is not added to docs/content/docs/connectors/table/formats/raw.md or the Chinese equivalent.

[spuru9]

nit: Pattern.quote(lineDelimiter) is recompiled on every message — since the delimiter is fixed, could this be pre-compiled as a Pattern field in the constructor?

[spuru9]

nit: delimiterBytes is allocated on every row — since both lineDelimiter and charsetName are fixed at construction time, this could be computed once and stored as a field.