[FLINK] Add RowKind column support for Gluten-Flink

[ggjh-159]

Description

Background

Follows up on the design note in apache/gluten#12164 (Gluten Flink Pending tasks), where retraction semantics is called out as a pending gap:

retraction semantics — this feature is required for group aggregate and join, which has not been implemented, maybe we should introduce the rowkind just as defined in flink, combine it with row vector.

Flink RowData carries a RowKind enum (+I / -U / +U / -D) that distinguishes changelog records — insert / update-before / update-after / delete. Gluten-Flink hands RowData to the velox C++ engine via the RowData → RowVector conversion path, but RowKind is dropped during that conversion today. As a result, the velox side sees every row as +I regardless of the original changelog marker.

Problem

• Sink operators (PrintSinkFactory, FileSystemSinkFactory, etc.) can't emit correct -U / +U / -D prefixes for upsert / delete streams — outputs are uniformly +I[...].
• Retract aggregations, temporal joins, and other row-kind-aware operators lose the row-kind signal once data crosses into velox.
• Nexmark / E2E tests over upsert streams produce incorrect changelog output.

Proposal

Encode RowKind as a synthetic column that flows through the entire plan and is interpreted at three layers: the Java ↔ C++ boundary, the velox stateful operators that must act on changelog semantics, and the sink layer that emits the changelog prefix.

• Type: TINYINT (Flink's RowKind enum ordinal: 0=INSERT, 1=UPDATE_BEFORE, 2=UPDATE_AFTER, 3=DELETE).
• Column name: reserved placeholder, invisible to user SQL($row_kind).
• Position: trailing column appended by the conversion layer.

The Row-kind-aware operators in velox (retract aggregation, temporal/regular join, top-N, deduplication, etc.) must read it to apply insert / update-before / update-after / delete semantics to their state. Stateful operators that ignore the column would either treat -U / -D as +I (incorrect accumulation) or drop them silently — both wrong.

Preserve the RowKind signal end-to-end across the gluten-flink three-layer architecture (Java / JNI / velox C++), and render the correct +I / -U / +U / -D prefix at the sink layer, establishing the data-carrier foundation for subsequent retract semantics

Design principles: do not pollute velox core data structures, do not pollute the Flink planner, zero intrusion on the Java side.

Task

• Foundation data structure: New ChangelogRowVector data structure (value + rowKind + appendOnly) + C++ UT
• Boundary data passing:
  • JNI boundary helpers: split merged RowVector into ChangelogRowVector on input; merge back to merged RowVector on output
  • Input boundary: RowData to merged RowVector conversion appends $row_kind column; Output boundary: merged RowVector to RowData conversion restores RowKind from the trailing column
• Compute framework compatibility: StreamRecord upgrade (carries ChangelogRowVector) + StatefulOperator base class refactor (split/recompose row kind on input/output, with a row-kind-rebuild virtual hook) + all call-site adaptations
• Operator adaptation:
  • Source/Calc/Sink Operator(Stateless)
  • agg/window/join ... (Stateful)

Gluten version

main branch

[Moemenmohamed24]

Hi , I would like to work on this improvement; would you grant me assign?

[ggjh-159]

Hi , I would like to work on this improvement; would you grant me assign?

Hi @Moemenmohamed24 . Discussions are underway regarding the best approach to adding a "row kind" column; some preliminary work has already been completed, and relevant pull requests will be released for review soon. Once the implementation strategy is finalized, the subsequent work—involving the modification of stateful operators and the implementation of necessary adaptations—will require broader participation.

[ggjh-159]

Hi, @lgbo-ustc @zhanglistar @KevinyhZou @Hw-TinY @Moemenmohamed24. This describes the changes in data format following the addition of RowKind. The MergedRowVector differs from the standard RowVector by including an extra $row_kind column at the end, while ChangelogRowVector is a new data structure defined in Velox that encapsulates the original RowVector along with the new $row_kind column.

• In Gluten-Flink, MergedRowVector is used for both input and output, and the component handles the conversion to and from RowData.
• Velox4j is responsible for converting between MergedRowVector and ChangelogRowVector.
• On the Velox side, the system processes StreamRecord objects that encapsulate a ChangelogRowVector; these records are passed along the operator chain, with individual operators selectively processing them based on their specific functions.
• Specifically, within the stateless Calc operator, the ChangelogRowVector must be converted into a MergedRowVector for processing by the Velox FilterProject operator (which performs filtering and projection); once the result is obtained, it is re-wrapped as a ChangelogRowVector for output.

The RowKind column is implicitly added to the operator chain, with conversion logic implemented in the Input/Output Bridge and Velox4J; the execution plan remains unaware of this column's existence.

Do you consider this an appropriate approach for adding RowKind column support to enable retraction semantics? Or are there other aspects that require further discussion?

[KevinyhZou]

Thanks for the detailed design. and I have some questions:

• 1. The flink rowdata convert to velox's ChangelogRowVector, does this means that：the input rowdata is generated by flink operators (like source), and then convert to velox data structure? as we have implemented some native operators(kafka/pulsar source) in velox side, should we use these to process the data directly without these convertions?
• 2. ChangeRowVector is a new data structure, can we use StatefulRecord that has been introduced in Velox by adding a new field RowKind instead of using this new one?

[ggjh-159]

Thanks for the detailed design. and I have some questions:

• 1. The flink rowdata convert to velox's ChangelogRowVector, does this means that：the input rowdata is generated by flink operators (like source), and then convert to velox data structure? as we have implemented some native operators(kafka/pulsar source) in velox side, should we use these to process the data directly without these convertions?
• 2. ChangeRowVector is a new data structure, can we use StatefulRecord that has been introduced in Velox by adding a new field RowKind instead of using this new one?

@KevinyhZou Thank you for your questions, here are the corresponding answers:
i. The data conversion here does not introduce a new path; instead, it adds a hidden column—$row_kind—to the existing data. The C++ native source should directly support constructing and outputting a ChangelogRowVector. In scenarios where C2R (Columnar-to-Row) or R2C (Row-to-Columnar) conversions are not required, data should flow continuously through the C++ operator chain without being converted to Flink RowData.
ii. Velox does not have a direct implementation of StatefulRecord. The existing StatefulRecord in velox4j corresponds to Java's StreamRecord, whereas the new ChangelogRowVector should correspond to GenericRowData.