[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:

• bb33242 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[rmetzger]

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

[featzhang]

Thanks for the thorough review @spuru9 and @rmetzger!

Here's what was addressed in this update:

1. Commit message: Updated to include the Jira ticket ID: [FLINK-39401][formats].

2. Round-trip compatibility fix (serializer/deserializer incompatibility): The deserializer now strips the single trailing empty string produced when a message ends with the delimiter. This makes serialization and deserialization fully round-trip compatible — serialize("hello") → "hello\n" → deserialize → ["hello"] (1 row, not 2). Added testDeserializeTrailingDelimiter_noExtraRow and testRoundTrip_serializeThenDeserialize tests to verify this.

3. Pre-compiled Pattern: Pattern.quote(lineDelimiter) is now pre-compiled once in the constructor as a Pattern lineDelimiterPattern field, eliminating repeated compilation on every message.

4. Pre-computed delimiterBytes: delimiterBytes is now computed once in the constructor and stored as a field, eliminating repeated allocation on every row serialization.

5. Documentation: Added raw.line-delimiter option to both docs/content/docs/connectors/table/formats/raw.md and docs/content.zh/docs/connectors/table/formats/raw.md, including a note on round-trip compatibility.

[spuru9]

Thank you for the change.
LGTM +1
@rmetzger Feel free to give a write approval or another review.