Skip to content

[GLUTEN-12459][FLINK] feat: supprt RowKind via $row_kind TINYINT column between Flink and Velox#12510

Open
ggjh-159 wants to merge 3 commits into
apache:mainfrom
ggjh-159:feat/support-rowkind
Open

[GLUTEN-12459][FLINK] feat: supprt RowKind via $row_kind TINYINT column between Flink and Velox#12510
ggjh-159 wants to merge 3 commits into
apache:mainfrom
ggjh-159:feat/support-rowkind

Conversation

@ggjh-159

Copy link
Copy Markdown
Contributor

Related issue: #12459

What changes are proposed in this pull request?

Flink RowData carries a RowKind header (+I / -U / +U / -D, byte ordinals 0/1/2/3) that today is lost the moment a row is converted to a velox RowVector. As a result every row exiting the velox pipeline is stamped +I, retract sinks print wrong prefixes, and any future retract-aware operator loses its input signal.

This PR threads RowKind through the conversion layer as a trailing TINYINT column named $row_kind, matching the convention already used on the velox side (kRowKindColumnName in velox/experimental/stateful/RowKind.h, ChangelogRowVector::toMergedRowVector). The user-visible schema is unchanged; the extra column is appended/stripped at the Flink boundary.

Changes:

  • ArrowVectorWriter: add TinyIntVectorWriter + TINYINT → ArrowType.Int(8, true) mapping so TINYINT columns can be written from RowData.getByte(...).
  • ArrowVectorAccessor: add TinyIntVectorAccessor so TINYINT columns can be read back as byte.
  • FlinkRowToVLVectorConvertor:
    • fromRowData now appends a single-row $row_kind TINYINT vector populated from row.getRowKind().toByteValue().
    • toRowData reads only rowType.size() user fields, then looks at the trailing column. If it is the $row_kind TINYINT vector, each row is reconstructed with GenericRowData.ofKind(...); otherwise the previous GenericRowData.of(...) path is taken (INSERT only), preserving backwards compatibility with append-only streams that never carry the column.

How was this patch tested?

  • UTs
  • E2E test, nexmark stateless sql

Depends on the velox4j(bigo-sg/velox4j#57).

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant