Skip to content

feat(amber): columnar hardening (versioning, compression, sink) #8567

Description

@Ma77Ball

Feature Summary

In one sentence: harden the columnar path for real use: version the wire format, add optional compression, offer a vectorized write-to-Iceberg sink, and speed up the row decode for the cases that still need rows.

Parent: #8556 (opt-in columnar execution). PR 4 (final) of the stacked series; pairs with #8561.

What is this?

The earlier PRs made columnar execution work. This one makes it safe to evolve and cheaper at the edges: a version stamp so old and new workers never misread each other, optional compression to shrink what crosses the network, a sink that writes Arrow straight to storage, and faster decode helpers for the fallback path.


Proposed Solution or Design

Wire versioning. ColumnarFrame now carries a format version. A worker that sees a version it does not support rejects it loudly instead of quietly misreading the bytes. This is the seatbelt that lets the format change later without silent corruption.

%%{init: {'theme':'dark', 'themeVariables': {'background':'#000000','lineColor':'#000000'}}}%%
flowchart TD
  IN[ColumnarFrame v=N] --> Q{N supported?}
  Q -->|yes| GO[read batch]
  Q -->|no| ERR[reject with a clear error]
  classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
  style GO stroke:#1B7F3B
  style ERR stroke:#B0451E
Loading

Optional wire compression. LZ4_FRAME or ZSTD via arrow-compression, config-gated. The reader auto-detects compressed versus uncompressed, so a compressed sender and a plain sender interoperate.

Vectorized Iceberg sink. IcebergTableWriter.writeArrowBatch writes an Arrow batch straight to Iceberg through a lazy ArrowRecordView that reads values from the Arrow columns, materializing no intermediate rows. Gated by enable-vectorized-sink.

%%{init: {'theme':'dark', 'themeVariables': {'background':'#000000','lineColor':'#0F766E'}}}%%
flowchart LR
  A[Arrow batch] --> V[ArrowRecordView: read straight from columns]
  V --> ICE[Iceberg data file]
  classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
Loading

Faster decode. Schema-once and projected decode helpers in ArrowUtils, so the fallback-to-rows path pays less.

Concern Before After
format change silent misread risk version stamp, explicit reject
network size uncompressed optional LZ4/ZSTD
sink write decode to rows first Arrow straight to Iceberg
decode resolve schema per row resolve once, decode only needed columns

Verified: batches round-trip unchanged including compressed and versioned frames; 2-worker shuffle row == columnar with the vectorized sink on.


High-level overview. Part of #8556.

Activity

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

Metadata

Metadata

Assignees

Labels

No labels
No labels

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions