featzhang opened a new pull request, #27897:
URL: https://github.com/apache/flink/pull/27897
## 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
```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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
