[GLUTEN-12426][FLINK] Feat: Add columnar StreamRecordTimestampInserter operator#12428
[GLUTEN-12426][FLINK] Feat: Add columnar StreamRecordTimestampInserter operator#12428ggjh-159 wants to merge 7 commits into
Conversation
ggjh-159
force-pushed
the
feat/stateful-stream-record-timestamp-inserter
branch
from
47c85b5 to
c2fd991
Compare
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| class GlutenRowtimeInserterHelperTest { | ||
|
|
There was a problem hiding this comment.
We use max timestamp in a batch record to replace the per-row timestamp, which is a different semantics，can we add a test for this？
There was a problem hiding this comment.
All three sink factories (Print, Fuzzer/Discard, FileSystem) wire requiresTimestamp = false, so the inserter is removed from the op chain and the batch-max path in GlutenRowtimeInserterHelper has no runtime caller in this PR. A test here would exercise a path no real sink reaches.
Batch-max computation itself is covered by velox C++ UT StreamRecordTimestampInserterTest.emitsBatchMaxTimestamp (PR bigo-sg/velox#64). The replacement/removal branches are covered by GlutenRowtimeInserterHelperTest.
| // inserter so the chain stays columnar end-to-end; otherwise this is a no-op. | ||
| Transformation<RowData> veloxFileWriterInput = | ||
| GlutenRowtimeInserterHelper.processTransformation( | ||
| (Transformation<RowData>) fileWriterTransformation.getInputs().get(0)); |
There was a problem hiding this comment.
In gluten-flink，operator StreamingFileWriter will totally offload to velox, and velox's file writer will not rely on the StreamRecord.timestamp to determine partition, roll file, or commit partition, therefore, StreamRecordTimestampInserter is just a useless operator, we should just remove it from the op chain.
There was a problem hiding this comment.
Agreed. Updated FileSystemSinkFactory to call GlutenRowtimeInserterHelper.processTransformation(input, /*requiresTimestamp=*/false), which removes any native StreamRecordTimestampInserter from the input chain instead of replacing it with the columnar one. Same treatment applied to PrintSinkFactory and FuzzerSourceSinkFactory.
…stamp-inserter' into feat/stateful-stream-record-timestamp-inserter
…m-record-timestamp-inserter # Conflicts: # .github/workflows/flink.yml # gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
2 participants
Issue: #12426
What changes are proposed in this pull request?
Adds a Gluten columnar replacement for Flink's row-mode
StreamRecordTimestampInserter, so the sink input chain stays columnar end-to-end when the sink is offloaded to Velox.The native Flink inserter (per-row timestamp) is replaced by
GlutenStreamRecordTimestampInserter, whose underlying Velox operator computes the batch-max timestamp and forwards it viaStatefulRecord.timestamp. The downstreamTimestampSettingStatefulRecordOutputBridgethen stamps the FlinkStreamRecordin Java without re-entering C++.Changes:
GlutenRowtimeInserterHelper— entry point invoked by each sink factory'sbuildVeloxSink; detects a native inserter on the input chain and rebuilds it as the Gluten columnar inserterGlutenStreamRecordTimestampInserter— new columnar operator (<StatefulRecord, StatefulRecord>)TimestampSettingStatefulRecordOutputBridge— collects withrecord.getTimestamp(), no JNI timestamp extractionPrintSinkFactory,FuzzerSourceSinkFactory,FileSystemSinkFactory— invoke the helper on their input streamsDepends on the corresponding velox PR <bigo-sg/velox#64> and velox4j PR <bigo-sg/velox4j#52>.
How was this patch tested?
GlutenRowtimeInserterHelperTest— 3 unit tests (non-OneInputTransformation no-op, non-inserter operator no-op, full replacement path):mvn -pl gluten-flink/ut test -Dtest=GlutenRowtimeInserterHelperTest