spuru9 commented on code in PR #27897:
URL: https://github.com/apache/flink/pull/27897#discussion_r3050820528
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java:
##########
@@ -63,7 +77,14 @@ public void open(InitializationContext context) throws
Exception {
@Override
public byte[] serialize(RowData row) {
try {
- return converter.convert(row);
+ byte[] valueBytes = converter.convert(row);
+ if (lineDelimiter == null || valueBytes == null) {
+ return valueBytes;
+ }
+ byte[] delimiterBytes =
lineDelimiter.getBytes(Charset.forName(charsetName));
Review Comment:
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.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java:
##########
@@ -72,6 +73,7 @@ public Set<ConfigOption<?>> optionalOptions() {
Set<ConfigOption<?>> options = new HashSet<>();
options.add(RawFormatOptions.ENDIANNESS);
options.add(RawFormatOptions.CHARSET);
+ options.add(RawFormatOptions.LINE_DELIMITER);
Review Comment:
raw.line-delimiter is not added to
docs/content/docs/connectors/table/formats/raw.md or the Chinese equivalent.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java:
##########
@@ -92,6 +108,34 @@ public RowData deserialize(byte[] message) throws
IOException {
return rowData;
}
+ @Override
+ public void deserialize(byte[] message, Collector<RowData> out) throws
IOException {
+ if (lineDelimiter == null) {
+ // no delimiter: default single-record behavior
+ RowData row = deserialize(message);
+ if (row != null) {
+ out.collect(row);
+ }
+ return;
+ }
+
+ if (message == null) {
+ return;
+ }
+
+ Charset charset = Charset.forName(charsetName);
+ String decoded = new String(message, charset);
+ String[] parts = decoded.split(Pattern.quote(lineDelimiter), -1);
Review Comment:
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?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java:
##########
@@ -63,7 +77,14 @@ public void open(InitializationContext context) throws
Exception {
@Override
public byte[] serialize(RowData row) {
try {
- return converter.convert(row);
+ byte[] valueBytes = converter.convert(row);
+ if (lineDelimiter == null || valueBytes == null) {
+ return valueBytes;
+ }
+ byte[] delimiterBytes =
lineDelimiter.getBytes(Charset.forName(charsetName));
Review Comment:
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.
--
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]
