[FLINK-39401] Extend raw format to support line-delimiter option by featzhang · Pull Request #27897 · apache/flink

featzhang · 2026-04-05T15:12:17Z

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 · 2026-04-05T15:15:37Z

CI report:

2742e22 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

rmetzger · 2026-04-08T11:41:18Z

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

featzhang · 2026-04-11T16:27:32Z

Thanks for the thorough review @spuru9 and @rmetzger!

Here's what was addressed in this update:

Commit message: Updated to include the Jira ticket ID: [FLINK-39401][formats].
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.
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.
Pre-computed delimiterBytes: delimiterBytes is now computed once in the constructor and stored as a field, eliminating repeated allocation on every row serialization.
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.

RocMarshal

Thanks @featzhang for the contribution.
Left some comments.
PTAL~

featzhang · 2026-04-16T01:53:10Z

Review Comments Addressed

✅ Fixed comments (from @RocMarshal)

#	File	Line	Fix	Status
1	RawFormatLineDelimiterTest.java	61	Use `StandardCharsets.UTF_8` instead of string literal `"UTF-8"`	✅
2	RawFormatLineDelimiterTest.java	63	Simplify assertion using `hasToString()`	✅
3	RawFormatDeserializationSchema.java	68	Add note that `lineDelimiter` and `lineDelimiterPattern` are either both null or both non-null	✅
4	RawFormatDeserializationSchema.java	131	Design explanation for null message handling (see below)	💬

📁 Modified files

RawFormatLineDelimiterTest.java: Added StandardCharsets import, replaced 7 string literals, simplified 1 assertion using hasToString()
RawFormatDeserializationSchema.java: Added nullability relationship comment for lineDelimiter and lineDelimiterPattern fields

💬 Design note for line 131 (null message with delimiter)

The current behavior (returning early on null message when a delimiter is set) is intentional and differs from the single-record path for the following reasons:

When lineDelimiter is set, the semantics shift from "one message → one row" to "one message → N rows via splitting". A null message in this context means there is no content to split, so emitting zero rows is the correct and expected result.
Emitting a row with a null field for a null message would be inconsistent with the splitting contract: if the delimiter is \n and the message is null, what does it mean to split null by \n? Producing a null row conflates "no data" with "one empty/null segment".
The existing test testDeserializeNullMessageWithDelimiter_noRows explicitly documents and verifies this design decision.

@RocMarshal PTAL, thanks!

featzhang · 2026-04-16T02:51:48Z

Thanks @RocMarshal for pointing that out! I've now gone through the entire RawFormatLineDelimiterTest.java file and applied the same fixes consistently:

Replaced all "UTF-8" string literals with StandardCharsets.UTF_8 (in .getBytes() calls) and StandardCharsets.UTF_8.name() (in constructor parameters)
Replaced all .getString(0).toString()).isEqualTo("...") assertions with .getString(0)).hasToString("...")
Added the missing import java.nio.charset.StandardCharsets;

The fix covers all test methods: testDeserializeWithoutDelimiter_singleRow, testDeserializeWithNewlineDelimiter_multipleRows, testDeserializeWithCustomMultiCharDelimiter, testDeserializeWithNullMessage_noOutput, testDeserializeWithGbkCharset, all serialization tests, testDeserializeTrailingDelimiter_noExtraRow, and testRoundTrip_serializeThenDeserialize.

@RocMarshal PTAL, thanks!

RocMarshal

Thanks @featzhang for update.
BTW, It would be perfect if the updated summary could reply point-by-point in the comments and be a bit more concise. That way, the reviewer wouldn’t need to break things down one by one and locate the corresponding changes to verify them.

LGTM+1 if the CI passed.

Looking forward to your next contribution~

featzhang · 2026-04-16T06:43:03Z

The CI failure in the latest build (#74134) is unrelated to this PR.

The failing tests are:

SqlGatewayOpenRestAPIDocsCompletenessITCase#testSqlGatewayRestApiDocsUpToDate
RuntimeOpenRestAPIDocsCompletenessITCase#testRuntimeRestApiDocsUpToDate

These tests check that rest_v1_sql_gateway.yml and rest_v1_dispatcher.yml are up-to-date. The failures are caused by recent changes on the master branch that modified REST API endpoints (e.g., [FLINK-38893], [FLINK-38894], [FLINK-38895]), which updated the generated REST docs after this PR's branch diverged. This is a known flaky issue in long-lived PRs and has nothing to do with the raw format changes in this PR.

The code changes in this PR only touch:

flink-table/flink-table-runtime module (raw format source files and tests)
docs/content (raw format documentation)

@flinkbot run azure

RocMarshal · 2026-04-16T10:49:03Z

Hi, @featzhang The CI failure was fixed in https://issues.apache.org/jira/browse/FLINK-39457.
Could you help rebase the latest master branch on the current path ?
Thx

…tion

@RocMarshal

…tion - Use StandardCharsets.UTF_8 instead of string literals in tests - Simplify assertions using hasToString() method - Add documentation for field nullability relationship Resolves review comments from @RocMarshal

…ut RawFormatLineDelimiterTest

featzhang · 2026-04-17T00:47:29Z

Hi, @featzhang The CI failure was fixed in https://issues.apache.org/jira/browse/FLINK-39457.

Could you help rebase the latest master branch on the current path ?

Thx

I have fixed it by rebase the latest main branch, and CI has been successful.

RocMarshal · 2026-04-17T01:42:11Z

Thanks @featzhang .
Merging...

snuyanzin · 2026-04-17T07:20:56Z

@featzhang , @RocMarshal any reason config option was added without FLIP/ML discussion?
asking since if we look at main FLIP page (section What is considered a "major change" that needs a FLIP?) https://cwiki.apache.org/confluence/display/FLINK/Flink+Improvement+Proposals

All of the following are public interfaces that people build around:

DataStream, DataSet, SQL and Table API, including classes related to that, such as StreamExecutionEnvironment

Classes marked with the @Public annotation

On-disk binary formats, such as checkpoints/savepoints

User-facing scripts/command-line tools, i.e. bin/flink, Yarn scripts, Kubernetes scripts

Configuration settings

Exposed monitoring information

featzhang · 2026-04-17T07:43:10Z

@featzhang , @RocMarshal any reason config option was added without FLIP/ML discussion? asking since if we look at main FLIP page (section What is considered a "major change" that needs a FLIP?) https://cwiki.apache.org/confluence/display/FLINK/Flink+Improvement+Proposals

All of the following are public interfaces that people build around:
DataStream, DataSet, SQL and Table API, including classes related to that, such as StreamExecutionEnvironment

Classes marked with the @Public annotation

On-disk binary formats, such as checkpoints/savepoints

User-facing scripts/command-line tools, i.e. bin/flink, Yarn scripts, Kubernetes scripts

Configuration settings

Exposed monitoring information

Hi @snuyanzin, thanks for raising this concern — and apologies for not initiating the discussion beforehand.

The raw.line-delimiter option is:

Optional with no default — existing behavior is entirely unchanged when the option is absent.
Additive-only — it adds a new capability without modifying any existing code paths.
Format-scoped — it is isolated to the raw format connector and does not touch any @Public API or SQL/Table API surface.

My understanding (which may be wrong) was that minor additive options within a format connector fall below the FLIP threshold, similar to how many other
format options (e.g., csv.field-delimiter, json.ignore-parse-errors) were introduced without a FLIP.

That said, I fully respect the community's position. If the consensus is that this option should have gone through FLIP/ML discussion, I'm happy to:

Open a FLIP retroactively to formally document the design rationale, or
Revert the change if the community feels it was premature.

What would you recommend, @snuyanzin and @RocMarshal?

RocMarshal · 2026-04-19T07:00:56Z

Thanks @snuyanzin for the attention and reminder, @featzhang for the response.

Please let me to add some information from my perspective at the time when reviewing the PR, and sorry for missing that earlier, which may have caused confusion:

I had also considered whether we needed a new FLIP to track this parameter. However, based on the following reasoning, I proceeded with the merge:

A: As @featzhang mentioned, this new parameter belongs to a sub-component or a secondary dimension.
B: Following from point A I looked for recent, similar changes ([1], [2]) and checked whether they introduced new FLIPs. Since none of them did, I referenced their development and review pattern, reviewed accordingly, and merged this PR.

That said, if the introduction of this parameter feels too abrupt, I’m happy to revert it immediately. We can then reintroduce this change later, strictly following the specification you outlined.

To ensure consistency, I've incorporated this into my review checklist. Going forward, I'll follow the specification closely to avoid similar issues.

Any input is appreciated!

[1]https://issues.apache.org/jira/browse/FLINK-38372
[2]

flink/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterOptions.java

Line 117 in ee827df

ConfigOptions.key("allowList")

snuyanzin · 2026-04-19T22:23:19Z

I don't think we need revert here however would be great to have more visibility
like ML or RN

RocMarshal · 2026-04-20T01:23:21Z

Thanks @snuyanzin .
That's reasonable.
I'll add it into the 2.3 release note.

featzhang changed the title ~~[FLINK] Extend raw format to support line-delimiter option~~ [FLINK-39401] Extend raw format to support line-delimiter option Apr 5, 2026

spuru9 reviewed Apr 8, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Apr 8, 2026

featzhang force-pushed the feature/raw-format-line-delimiter branch from 8205f0c to 3161013 Compare April 11, 2026 16:27

featzhang force-pushed the feature/raw-format-line-delimiter branch from 3161013 to bb33242 Compare April 12, 2026 09:38

spuru9 approved these changes Apr 12, 2026

View reviewed changes

RocMarshal requested changes Apr 15, 2026

View reviewed changes

RocMarshal self-assigned this Apr 15, 2026

featzhang force-pushed the feature/raw-format-line-delimiter branch from bb33242 to b219d10 Compare April 16, 2026 01:52

RocMarshal reviewed Apr 16, 2026

View reviewed changes

Comment thread ...ble-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java

Comment thread ...ble-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java Outdated

RocMarshal approved these changes Apr 16, 2026

View reviewed changes

Comment thread ...ble-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java

featzhang added 4 commits April 16, 2026 20:55

[FLINK-39401][formats] Extend raw format to support line-delimiter op…

dd362f1

…tion

Address review: use StandardCharsets.UTF_8 and hasToString() througho…

f62196d

…ut RawFormatLineDelimiterTest

[FLINK-39401][formats] Fix code formatting via spotless:apply

2742e22

featzhang force-pushed the feature/raw-format-line-delimiter branch from 8f7dea3 to 2742e22 Compare April 16, 2026 12:56

RocMarshal merged commit 2379582 into apache:master Apr 17, 2026

featzhang deleted the feature/raw-format-line-delimiter branch April 17, 2026 03:18

Conversation

featzhang commented Apr 5, 2026

Summary

Example SQL

Changes

Test Plan

Uh oh!

flinkbot commented Apr 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

rmetzger commented Apr 8, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

featzhang commented Apr 11, 2026

Uh oh!

spuru9 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RocMarshal left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

featzhang commented Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review Comments Addressed

✅ Fixed comments (from @RocMarshal)

📁 Modified files

💬 Design note for line 131 (null message with delimiter)

Uh oh!

Uh oh!

Uh oh!

featzhang commented Apr 16, 2026

Uh oh!

RocMarshal left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

featzhang commented Apr 16, 2026

Uh oh!

RocMarshal commented Apr 16, 2026

Uh oh!

featzhang commented Apr 17, 2026

Uh oh!

RocMarshal commented Apr 17, 2026

Uh oh!

snuyanzin commented Apr 17, 2026

Uh oh!

featzhang commented Apr 17, 2026

Uh oh!

RocMarshal commented Apr 19, 2026

Uh oh!

snuyanzin commented Apr 19, 2026

Uh oh!

RocMarshal commented Apr 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

flinkbot commented Apr 5, 2026 •

edited

Loading

spuru9 left a comment •

edited

Loading

featzhang commented Apr 16, 2026 •

edited

Loading