Skip to content

feat: Add ChangelogRowVector for RowKind propagation#70

Open
ggjh-159 wants to merge 5 commits into
bigo-sg:gluten-0530from
ggjh-159:feat/support-rowkind
Open

feat: Add ChangelogRowVector for RowKind propagation#70
ggjh-159 wants to merge 5 commits into
bigo-sg:gluten-0530from
ggjh-159:feat/support-rowkind

Conversation

@ggjh-159

Copy link
Copy Markdown

What changes are proposed in this pull request?

This introduces a new data structure ChangelogRowVector under velox/experimental/stateful/, intended as the standard data carrier for changelog-aware streaming between stateful operators.

It pairs an existing RowVector (schema columns) with a per-row RowKind vector (TINYINT, ordinal values aligned with Flink's
org.apache.flink.types.RowKind: 0=INSERT, 1=UPDATE_BEFORE, 2=UPDATE_AFTER, 3=DELETE).

The new type lives next to StreamRecord and does not modify any existing operator, vector, or plan node. This PR only adds the data-carrier primitive + C++ unit tests. Wiring it into the operator chain is tracked as a follow-up).

Public API

namespace facebook::velox::stateful {

class ChangelogRowVector {
 public:
  // rowKind may be null (appendOnly, all INSERT) or non-null with size == value->size().
  explicit ChangelogRowVector(RowVectorPtr value,
                              SimpleVectorPtr<int8_t> rowKind = nullptr);

  const RowVectorPtr& value() const;
  const SimpleVectorPtr<int8_t>& rowKind() const;  // nullable
  vector_size_t size() const;
  bool appendOnly() const;

  // Split/merge helpers for the JNI boundary.
  // `$row_kind` must be the trailing column of the merged RowVector.
  static ChangelogRowVectorPtr fromMergedRowVector(RowVectorPtr merged);
  RowVectorPtr toMergedRowVector() const;
};

}  // namespace facebook::velox::stateful

Key design decisions

  • appendOnly is inferred, not a parameter. True when rowKind is null or a Constant(INSERT) vector. This keeps the constructor surface minimal and avoids two callers disagreeing on the flag.
  • RowKind is not added to RowVector schema. It travels alongside the data so existing velox operators (FilterProject, HashAggregation, etc.) remain byte-for-byte unchanged.
  • $row_kind is constrained to the trailing column. Lets fromMergedRowVector split without scanning names, and lets
    toMergedRowVector round-trip exactly.

Why are the changes needed?

Flink RowData carries a per-row RowKind that distinguishes +I / -U / +U / -D. When changelog streams cross into velox
through the stateful operator chain, this signal is currently dropped at the input boundary, which silently corrupts retract semantics (group aggregate, temporal join, sink output prefixes).

This PR is the foundation layer of a multi-PR effort to restore that signal end-to-end. Centralizing the (value, rowKind, appendOnly) binding in a single primitive lets downstream operators compose with it instead of each call site re-deriving the appendOnly flag.

Related design + phased plan: support-rowkind-column-issue

Test

New C++ unit test velox_stateful_changelog_row_vector_test

@ggjh-159 ggjh-159 changed the title feat(stateful): add ChangelogRowVector for RowKind propagation [FLINK] Feature: add ChangelogRowVector for RowKind propagation Jul 10, 2026
@ggjh-159 ggjh-159 marked this pull request as draft July 13, 2026 06:59
@ggjh-159 ggjh-159 changed the title [FLINK] Feature: add ChangelogRowVector for RowKind propagation feat: add ChangelogRowVector for RowKind propagation Jul 13, 2026
@ggjh-159 ggjh-159 marked this pull request as ready for review July 13, 2026 07:49
@ggjh-159 ggjh-159 changed the title feat: add ChangelogRowVector for RowKind propagation feat: Add ChangelogRowVector for RowKind propagation Jul 13, 2026
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.

1 participant