feat: Added IcebergArrowInputSourceReader by Shekharrajak · Pull Request #19510 · apache/druid

Shekharrajak · 2026-05-23T09:49:54Z

Fixes #19498

Description

Added IcebergArrowInputSourceReader using iceberg-arrow vectorized API
Returns the live Table object so the Arrow reader can drive scan planning
Two new optional JSON properties on the input spec. useArrowReader=false keeps existing behaviour byte-for-byte; useArrowReader=true switches to the Arrow path. arrowBatchSize defaults to 1024.

Internals: IcebergArrowInputSource loads the Iceberg Table, builds a filtered/projected TableScan, uses ArrowReader to produce ColumnarBatches, converts Arrow vectors into Druid InputRows, and tracks processed bytes from Arrow buffers.

Tests cover Arrow reader behavior, projection pruning, residual filter handling, null InputStats, and non-splittable Arrow ingestion behavior.

Release note

Apache Iceberg ingestion now supports an opt-in vectorized reader path backed by iceberg-arrow. Enable by setting "useArrowReader": true on the iceberg input source. The Arrow path automatically applies V2 delete files (positional and equality), handles schema evolution, pushes column projection and predicates into the scan planner, and is 2x–3x faster than the existing path.
Future Iceberg spec features (V3 deletion vectors, row lineage, V4+) become available on Iceberg version bumps with no Druid code changes. Default remains the existing path; both coexist.

This PR has:

Shekharrajak · 2026-05-23T09:50:25Z

Benchmark :

| numRows  | numCols | icebergArrowInputSourceReader    | icebergInputSourceReader         | Speedup |
|         |         |  ms/op         throughput        |  ms/op         throughput        | (vs baseline) |
+---------+---------+----------------------------------+----------------------------------+---------+
| 100,000 |   5     |   17.70   5,651,074 rows/s       |   51.35   1,947,542 rows/s       |  2.90x  |
| 100,000 |  15     |   44.11   2,266,881 rows/s       |   95.81   1,043,680 rows/s       |  2.17x  |
| 500,000 |   5     |   74.92   6,673,450 rows/s       |  187.73   2,663,420 rows/s       |  2.51x  |
| 500,000 |  15     |  197.97   2,525,637 rows/s       |  420.94   1,187,817 rows/s       |  2.13x  |
================================================================================
Summary:
  Best speedup:    2.90x  (numRows=100,000, numCols=5)
  Worst speedup:   2.13x  (numRows=500,000, numCols=15)
  Geomean speedup: 2.41x

Shekharrajak · 2026-05-23T09:51:06Z

Looking into benchmark module which will be used throughout the arrow and datafusion integration to quickly check the improvements.

Shekharrajak · 2026-05-23T12:23:16Z

+  private final boolean useArrowReader;
+
+  @JsonProperty
+  private final int arrowBatchSize;


With a batch of 1024 values in a contiguous buffer -> emit SIMD instructions that process 8-16 rows per CPU cycle.

Decompression amortization -> batch-read helps in decompress once and consume many row, since parquet compressed pages .

Shekharrajak · 2026-05-23T12:50:25Z

ingestion time at 100k × 5

Read

Arrow implementation : 17ms
Current implementaiton : 53ms

Index + Persist (= total − read)

~330ms
Similar time on both implementation

total time

Arrow implementation: 347 ms
current implementation: 410 ms

Index+persist does substantially more work per row than read. So even though Arrow makes read ~3x faster, that gain is dwarfed when amortized over the much-larger indexing cost.

Benchmark added https://github.com/Shekharrajak/druid/pull/1/changes

Shekharrajak · 2026-05-23T13:07:52Z

looks like network error https://github.com/apache/druid/actions/runs/26333466916/job/77523245546?pr=19510 - please trigger the CI check again.

FrankChen021

Severity	Findings
P0	0
P1	2
P2	1
P3	0
Total	3

Reviewed 7 of 7 changed files.

This is an automated review by Codex GPT-5.5

…ccess

…d API

…st constructor arity

…t line breaking, DateTimes.utc, Maps.newHashMapWithExpectedSize)

…Allocation init failure

…ly failing)

…aInputSource

…=true

Shekharrajak · 2026-05-24T17:27:18Z

  public int estimateNumSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec) throws IOException
  {
+    if (useArrowReader) {
+      return 1;


Future PR: Enable parallel ingestion (maxNumConcurrentSubTasks > 1) when useArrowReader: true by adding split-coordination plumbing to IcebergInputSource. Currently Arrow mode forces a single subtask to guarantee correctness;

FrankChen021

I reviewed the follow-up changes. The projection issue and Arrow split handling thread look resolved; I left a separate inline reply on the remaining residual-filter snapshot-time parity gap.

Reviewed 7 of 7 changed files.

This is an automated review by Codex GPT-5.5

…lready opens java.nio

Shekharrajak · 2026-05-26T09:55:16Z

+{
+  // Pin Arrow to Unsafe allocator: Netty backend fails on JDK 25 (EmptyByteBuf.memoryAddress UnsupportedOperationException).
+  static {
+    if (System.getProperty("arrow.allocation.manager.type") == null) {


NettyAllocationManager. throws UnsupportedOperationException on JDK 25 (incompatible with newer JDK module encapsulation). Pinning to "Unsafe" uses sun.misc.Unsafe which works on all supported JDKs (21, 25).

CI checks were failing .

Shekharrajak · 2026-05-26T09:55:27Z

+ * Column projection and predicate push-down are applied at scan planning time so only requested
+ * columns and matching files are read from storage.
+ *
+ * Note: iceberg-arrow currently supports Parquet data files only. ORC and Avro files will throw


Supports only parquet data file

Shekharrajak · 2026-05-26T09:56:11Z

  private final ResidualFilterMode residualFilterMode;

+  @JsonProperty
+  private final boolean useArrowReader;


Adds useArrowReader opt-in flag

FrankChen021

I reviewed the follow-up changes for correctness, edge cases, concurrency, and integration risks; no new issues found.

Reviewed 7 of 7 changed files.

This is an automated review by Codex GPT-5.5

Shekharrajak · 2026-05-27T14:21:20Z

Hi @jtuglu1 @clintropolis @FrankChen021 - Please have a look, I have benchmark PR drafted which we will use across all features to see the improvements we are getting through vectorization & arrow implementations.

cecemei

The benchmark results look good! I don't see any similar benchmark coverage in this area, but it's definitely worth adding.

…(type iceberg_arrow), extract AbstractIcebergInputSource

…residual-FAIL

+      @JsonProperty("residualFilterMode") @Nullable ResidualFilterMode residualFilterMode,
+      // deprecated: use type "iceberg_arrow" instead. retained for spec back-compat.
+      @JsonProperty("useArrowReader") @Nullable Boolean useArrowReader,
+      @JsonProperty("arrowBatchSize") @Nullable Integer arrowBatchSize


FrankChen021

Severity	Findings
P0	0
P1	0
P2	2
P3	0
Total	2

Reviewed 11 of 11 changed files.

This is an automated review by Codex GPT-5.5

Shekharrajak · 2026-06-05T16:55:36Z

Flaky test found : #19563

FrankChen021

Reviewed 12 of 12 changed files.

The follow-up updates handle both prior review threads: the Arrow/Iceberg Arrow license metadata is now registered, with arrow-memory-netty excluded, and residual enforcement now applies snapshotTime before planning. I did not find any new high-confidence PR-caused issues in the updated diff.

This is an automated review by Codex GPT-5.5

FrankChen021

I have reviewed the code for correctness, edge cases, concurrency, and integration risks; no issues found.

Reviewed 12 of 12 changed files.

This is an automated review by Codex GPT-5.5

Shekharrajak · 2026-06-15T16:51:28Z

Thanks @FrankChen021 , can we go ahead and make this foundation merged in. I have next phases planned as part of vision #19456

kfaraz · 2026-08-01T14:56:17Z

Removing from Druid 38 milestone since this PR has merge conflicts and needs an approval.

Shekharrajak · 2026-08-03T07:42:36Z

Updated the branch with master.

FrankChen021

Severity	Findings
P0	0
P1	1
P2	1
P3	0
Total	2

Reviewed 12 of 12 changed files. The upstream merge does not change the PR-authored Arrow path, but the current diff still corrupts dictionary-backed values and cannot support the delete files it claims to handle.

This is an automated review by Codex GPT-5.6-Sol

FrankChen021 · 2026-08-03T12:26:42Z

+    final int numCols = batch.numCols();
+    final Map<String, Object> event = Maps.newHashMapWithExpectedSize(numCols);
+    for (int col = 0; col < numCols; col++) {
+      final FieldVector vec = batch.column(col).getFieldVector();


[P1] Decode dictionary-backed vectors before extraction

ColumnVector.getFieldVector() explicitly returns a potentially dictionary-encoded vector. For dictionary-backed strings or partition constants this is an IntVector, so extractValue ingests the dictionary ID instead of the actual value, silently corrupting dimensions. Use getArrowVector() or the typed ColumnVector accessors and assert actual string or partition values in tests; testWithEqualsFilter currently avoids this assertion.

FrankChen021 · 2026-08-03T12:26:42Z

+/**
+ * Reads an Iceberg table via iceberg-arrow's {@link ArrowReader}, yielding {@link InputRow} objects.
+ *
+ * Delete application (V2 equality and positional deletes), type coercion, and schema evolution are


[P2] Do not claim unsupported delete-file handling

The class claims equality and positional deletes are applied, but Iceberg 1.10 documents that ArrowReader.open requires no delete files and throws UnsupportedOperationException otherwise. V2 snapshots containing deletes therefore fail ingestion. Detect and fall back to a capable path, or reject and document this limitation explicitly; add coverage for both delete types.

github-actions Bot added the Area - Dependencies label May 23, 2026

Shekharrajak force-pushed the feature/iceberg-arrow-reader branch from 6297e2f to 9da5049 Compare May 23, 2026 10:24

Shekharrajak mentioned this pull request May 23, 2026

iceberg arrow reader harness Shekharrajak/druid#1

Open

Shekharrajak commented May 23, 2026

View reviewed changes

Shekharrajak mentioned this pull request May 23, 2026

[Proposal] Native, vectorised, zero-copy execution path for Druid #19456

Open

FrankChen021 reviewed May 24, 2026

View reviewed changes

Shekharrajak added 7 commits May 24, 2026 20:28

deps: add arrow 15.0.2 and iceberg-arrow to pom dependency management

823018b

feat: add retrieveTable() to IcebergCatalog for direct Table object a…

a873c4d

…ccess

feat: add IcebergArrowInputSourceReader using iceberg-arrow vectorize…

c420836

…d API

feat: wire useArrowReader + arrowBatchSize into IcebergInputSource

866a093

test: add IcebergArrowInputSourceReaderTest; fix IcebergInputSourceTe…

2339330

…st constructor arity

style: fix checkstyle and forbidden-apis violations (imports, argumen…

8447ce0

…t line breaking, DateTimes.utc, Maps.newHashMapWithExpectedSize)

fix: switch arrow-memory-netty to arrow-memory-unsafe to fix CI Arrow…

f151f5b

…Allocation init failure

Shekharrajak force-pushed the feature/iceberg-arrow-reader branch from 8808ec4 to f151f5b Compare May 24, 2026 14:59

Shekharrajak added 8 commits May 24, 2026 21:33

test: add regression for aggregator source column projection (current…

87bdc7f

…ly failing)

fix: drive Iceberg scan projection from ColumnsFilter, mirroring Delt…

c774363

…aInputSource

test: ColumnsFilter exclusion prunes unused columns at Iceberg scan

6db57a7

style: drop redundant comments per AGENTS.md hygiene

a7fa10c

test: regression for residual FAIL mode bypassed by Arrow path

113948f

fix: enforce residualFilterMode in Arrow reader path

6a3ad85

test: regression for parallel ingestion bypassing Arrow reader

ed6c2ff

fix: route splittable contract through Arrow path when useArrowReader…

498daa6

…=true

Shekharrajak commented May 24, 2026

View reviewed changes

fix(iceberg-arrow): open jdk.internal.misc for Arrow allocator on JDK 21

484ebea

FrankChen021 reviewed May 25, 2026

View reviewed changes

fix(iceberg-arrow): drop module surefire override; root pom argLine a…

bcbfe67

…lready opens java.nio

Shekharrajak commented May 26, 2026

View reviewed changes

FrankChen021 reviewed May 26, 2026

View reviewed changes

cecemei reviewed May 29, 2026

View reviewed changes

Comment thread ...ruid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergInputSource.java

Comment thread ...g-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSourceReader.java Outdated

Shekharrajak added 5 commits June 2, 2026 23:05

iceberg: split useArrowReader into dedicated IcebergArrowInputSource …

bf71afb

…(type iceberg_arrow), extract AbstractIcebergInputSource

iceberg: add IcebergArrowInputSourceTest covering non-splittable and …

3fbb73f

…residual-FAIL

iceberg: drop unused InputRowSchema import in IcebergInputSourceTest

2845d46

iceberg-arrow: null-guard InputStats in read() per nullable contract

d046573

iceberg-arrow: regression test for null InputStats via no-arg read()

197afce

github-advanced-security AI found potential problems Jun 2, 2026

View reviewed changes

FrankChen021 reviewed Jun 3, 2026

View reviewed changes

Comment thread ...iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSource.java

Comment thread extensions-contrib/druid-iceberg-extensions/pom.xml

Shekharrajak added 2 commits June 4, 2026 21:43

Fix Iceberg Arrow residual snapshot

6c37e3c

Register Iceberg Arrow licenses

7a467f8

FrankChen021 reviewed Jun 7, 2026

View reviewed changes

Shekharrajak mentioned this pull request Jun 10, 2026

fix: flaky Consul embedded test startup #19564

Merged

10 tasks

Shekharrajak added 2 commits June 11, 2026 22:02

Merge upstream master

df75d6d

Merge origin feature branch

fd9fe66

FrankChen021 reviewed Jun 15, 2026

View reviewed changes

jtuglu1 added this to the 38.0.0 milestone Jun 18, 2026

jtuglu1 self-requested a review June 23, 2026 07:58

kfaraz modified the milestones: 38.0.0, 39.0.0 Aug 1, 2026

Merge upstream master

d7a4e63

FrankChen021 reviewed Aug 3, 2026

View reviewed changes

Uh oh!

Conversation

Shekharrajak commented May 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Release note

Uh oh!

Shekharrajak commented May 23, 2026

Uh oh!

Shekharrajak commented May 23, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Shekharrajak commented May 23, 2026

Read

Index + Persist (= total − read)

total time

Uh oh!

Shekharrajak commented May 23, 2026

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Shekharrajak commented May 27, 2026

Uh oh!

cecemei left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Shekharrajak commented Jun 5, 2026

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Shekharrajak commented Jun 15, 2026

Uh oh!

kfaraz commented Aug 1, 2026

Uh oh!

Shekharrajak commented Aug 3, 2026

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Shekharrajak commented May 23, 2026 •

edited

Loading