Skip to content

feat: Add StreamRecordTimestampInserter operator#64

Merged
lgbo-ustc merged 5 commits into
bigo-sg:gluten-0530from
ggjh-159:feat/stateful-stream-record-timestamp-inserter
Jul 14, 2026
Merged

feat: Add StreamRecordTimestampInserter operator#64
lgbo-ustc merged 5 commits into
bigo-sg:gluten-0530from
ggjh-159:feat/stateful-stream-record-timestamp-inserter

Conversation

@ggjh-159

@ggjh-159 ggjh-159 commented Jul 2, 2026

Copy link
Copy Markdown

Issue: gluten#12426

Summary

Adds a new stateful operator StreamRecordTimestampInserter to velox/experimental/stateful/, the native counterpart of Flink's org.apache.flink.table.runtime.operators.sink.StreamRecordTimestampInserter.

Unlike the row-based Flink original (which reads the rowtime column row-by-row and stamps each StreamRecord), this operator works in columnar batches:

  • Reads the rowtime column of each input batch
  • Computes the batch's max timestamp (in epoch millis)
  • Emits the original RowVector wrapped in a StreamRecord carrying that timestamp

No watermark generation, no batch splitting — RowVector passes through end-to-end. Operator logic and plan node live entirely in C++.

Changes

  • New velox/experimental/stateful/StreamRecordTimestampInserter.{h,cpp} — operator implementation.
  • Modified velox/experimental/stateful/StatefulPlanNode.{h,cpp} — adds StreamRecordTimestampInserterNode PlanNode with project + rowtimeFieldIndex fields, registers serde ("StreamRecordTimestampInserterNode").
  • Modified velox/experimental/stateful/StatefulPlanner.{h,cpp} — adds dispatch + transformStreamRecordTimestampInserterOperator.
  • Modified velox/experimental/stateful/CMakeLists.txt — adds StreamRecordTimestampInserter.cpp to velox_stateful_exec.
  • New velox/experimental/stateful/tests/StreamRecordTimestampInserterTest.cpp + CMakeLists entry — unit tests.

Test plan

  • UTs.
  • End-to-end nexmark q0 validation.

1,
"Expected single-column timestamp vector from wrapped FilterProject");
auto child = timestampVector->childAt(0);
auto tsVector = child->as<SimpleVector<Timestamp>>();

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The result of watermark::getTimestampVector(...) should be SimpleVector<int64_t> ? and this can be seen from WatermarkGenerator and WatermarkAssigner

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In Flink file org/apache/flink/table/runtime/operators/sink/StreamRecordTimestampInserter.java

    @Override
    public void processElement(StreamRecord<RowData> element) throws Exception {
        final RowData rowData = element.getValue();
        // timestamp might be TIMESTAMP or TIMESTAMP_LTZ
        final long rowtime = rowData.getTimestamp(rowtimeIndex, precision).getMillisecond();
        element.setTimestamp(rowtime);
        output.collect(element);
    }

The type of the column vector at index rowtimeIndex should be Timestamp.

}

pushOutput(std::make_shared<StreamRecord>(
getPlanNodeId(), std::move(input_), maxTs));

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if input.size == 0, then maxTs = std::numeric_limits<int64_t>::min(), StreamRecord.hasTImestamp should be false

@ggjh-159 ggjh-159 requested a review from KevinyhZou July 6, 2026 06:39

@KevinyhZou KevinyhZou left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@lgbo-ustc

Copy link
Copy Markdown

format the code

@ggjh-159 ggjh-159 changed the title [FLNK] feat: add StreamRecordTimestampInserter operator feat: add StreamRecordTimestampInserter operator Jul 13, 2026
@ggjh-159 ggjh-159 changed the title feat: add StreamRecordTimestampInserter operator feat: Add StreamRecordTimestampInserter operator Jul 13, 2026
@ggjh-159

Copy link
Copy Markdown
Author

format the code

@lgbo-ustc cc

@lgbo-ustc lgbo-ustc merged commit cad8903 into bigo-sg:gluten-0530 Jul 14, 2026
10 of 13 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants