feat: Support V2 Format for Iceberg Ingestion by a2l007 · Pull Request #19534 · apache/druid

a2l007 · 2026-06-01T02:24:27Z

Fixes #19190

Description

This PR fixes Iceberg delete semantics which was previously broken for v2 tables. No spec changes are required. The same ingestion spec works for V1 and V2 tables. Delete detection is automatic at planning time.

In the V2 path, each FileScanTask is encoded as an IcebergFileTaskInputSource , containing the data file path, format, size, delete file metadata (paths, content type, equality field IDs), and the table schema captured as JSON at planning time.

IcebergNativeRecordReader processes each split:

Delete metadata is loaded upfront. For each associated delete file is processed based on whether it is a position delete file or an equality delete file.
- Position delete files: Rows to skip are read and filtered to only positions matching the current data file path.
- Equality delete files : Rows of key tuples are read per equality delete file.
The data file is streamed lazily using Iceberg's Parquet.read() builder with GenericParquetReaders.
Deletes are applied per record: each record's position is checked against the position delete set; its field values are checked against each equality delete set. Matching records are skipped.
Surviving records are converted to MapBasedInputRow via IcebergRecordConverter

FileIO is resolved per path: local paths use WarehouseFileIO, non-local paths (HDFS, S3) fall back to catalog.loadTable().io() using the namespace and table name passed through the split.

Release note

The druid-iceberg-extensions input source now correctly handles Iceberg V2 tables that contain row-level delete files. Previously, position deletes and equality deletes were silently ignored, causing deleted rows to appear in ingested segments. Druid now auto-detects delete files at planning time and applies them during ingestion with no spec changes required.

Key changed/added classes in this PR

IcebergCatalog
IcebergInputSource

This PR has:

been self-reviewed.
added documentation for new or modified features or behaviors.
a release note entry in the PR description.
added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
been tested in a test Druid cluster.

FrankChen021

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

Reviewed 8 of 8 changed files.

This is an automated review by Codex GPT-5.5

FrankChen021 · 2026-06-01T14:28:29Z

+{
+  private static final Logger log = new Logger(IcebergNativeRecordReader.class);
+
+  private static final WarehouseFileIO WAREHOUSE_FILE_IO = new WarehouseFileIO();


[P1] Undefined FileIO class breaks compilation

The new reader instantiates WarehouseFileIO, but no class with that name is defined in this package or elsewhere in the repo, and there is no import that could resolve it. Any build that compiles this extension will fail before the V2 ingestion path can run. Use an existing Iceberg FileIO implementation or add the missing implementation.

FrankChen021 · 2026-06-01T14:28:29Z


-      // Handle residual filter based on mode
-      if (detectedResidual != null) {
+      if (detectedResidual == null) {


[P2] Residual-filter check is inverted

This condition now enters the residual-handling branch when detectedResidual is null, even though null means no residual was found. With ResidualFilterMode.FAIL, valid scans that have no residual, such as partition-pruned scans, will throw a residual-filter error, while scans that do have a residual skip the warning/failure path. Restore the check to run only when detectedResidual != null.

FrankChen021

🟡 Changes recommended

The current V2 reader has one blocking correctness issue and two additional behavior regressions: nested equality deletes are not resolved through Iceberg's nested field names, fixed-width binary equality keys cannot match because byte arrays use identity equality, and the native V2 path bypasses the configured Parquet input format. The two earlier MergeLens findings on the missing FileIO class and inverted residual-filter check are resolved in the current head.

Reviewed 10 of 10 changed files.

Validation: static review of the incremental diff, full merge-base diff, current surrounding code, Druid input-source/task wiring, and the Iceberg 1.11.0 local sources; git diff --check passed. No builds, tests, dependency installs, or formatters were run.

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

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

FrankChen021 · 2026-09-20T12:45:11Z

+      for (String fieldName : fieldNames) {
+        key.put(fieldName, dataRecord.getField(fieldName));
+      }
+      return keys.contains(key);


[P2] Fixed binary equality keys use identity comparison

Finding: GenericParquetReaders.FixedReader returns a new byte[] for each fixed-width value. The delete key map and the data-row key map therefore contain distinct arrays, and HashMap.equals compares those arrays by identity rather than contents; equality deletes on Iceberg fixed fields never match and the deleted rows remain visible.

Suggestion: Canonicalize binary and fixed values to a content-based representation, or use deep/content equality when constructing and checking equality-delete keys.

FrankChen021

🟡 Changes recommended

The current V2 delete-file path still has two concrete compatibility regressions: it bypasses the configured Parquet InputFormat, and its native row conversion does not apply Druid dimension discovery/include-all/exclusion semantics. Both can change ingestion results for otherwise valid configurations.

The earlier findings for nested equality deletes and fixed-width binary equality keys appear resolved in the current head. The prior FileIO and residual-filter findings also remain resolved.

Reviewed 10 of 10 changed files.

Scope: updated-since-review incremental diff from 5fc1408, the full merge-base diff, and relevant Druid/Iceberg surrounding code.

Validation: static review only; git diff --check passed. No builds, tests, dependency installs, or formatters were run.

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

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

# Conflicts: # docs/development/extensions-contrib/iceberg.md # extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergCatalog.java # extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergInputSource.java

+            List<Integer> fieldIds = fieldIdsStr.isEmpty()
+                                     ? Collections.emptyList()
+                                     : Arrays.stream(fieldIdsStr.split(","))
+                                             .map(Integer::parseInt)


FrankChen021

🟡 Changes recommended

The current head retains two correctness regressions in the Iceberg V2 delete-aware path: it bypasses configured Parquet InputFormat behavior, and its native row conversion does not honor Druid dimension discovery/include-all/exclusion semantics. I also found that the new scan-planning path leaves Iceberg's CloseableIterable unclosed, which skips scan completion and can leak planning resources on failure.

The earlier findings for nested equality deletes and fixed-width binary equality keys appear resolved in the current head. The prior FileIO and residual-filter findings also remain resolved.

Reviewed 10 of 10 changed files:

docs/development/extensions-contrib/iceberg.md
extensions-contrib/druid-iceberg-extensions/pom.xml
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/common/IcebergDruidModule.java
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergCatalog.java
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergFileTaskInputSource.java
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergInputSource.java
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergNativeRecordReader.java
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergRecordConverter.java
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/WarehouseFileIO.java
extensions-contrib/druid-iceberg-extensions/src/test/java/org/apache/druid/iceberg/input/IcebergInputSourceTest.java

Scope: started from the prepared incremental diff, then reviewed the full merge-base diff, prior review findings, current-head code, Druid input-source/task wiring, and relevant Iceberg local sources.

Validation: static review only; git diff --check passed. No builds, tests, dependency installs, or formatters were run.

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

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

FrankChen021 · 2026-09-23T13:03:19Z

+      }
+      if (hasDeleteFiles) {
+        // V2 path: create a combined reader across all file-scan tasks.
+        final List<IcebergNativeRecordReader> taskReaders = v2Tasks.stream()


[P2] V2 reads ignore the configured input format

Finding: When any planned task has delete files, this branch routes every task through IcebergNativeRecordReader, but the reader is constructed only from InputRowSchema and the supplied InputFormat is discarded. ParquetInputFormat options such as flattenSpec and binaryAsString are therefore skipped for V2 tables, including data files without deletes, so a valid ingestion spec can produce different or missing indexed values from the existing warehouse-reader path.

Suggestion: Preserve the configured Parquet input-format semantics on the delete-aware path, either by carrying the relevant options into the native reader or by applying delete filtering around the existing Druid input-format reader.

FrankChen021 · 2026-09-23T13:03:19Z

+  {
+    DateTime timestamp = inputRowSchema.getTimestampSpec().extractTimestamp(row);
+
+    List<String> dimensions = inputRowSchema.getDimensionsSpec().getDimensionNames();


[P2] Native conversion ignores dimension discovery rules

Finding: The native V2 conversion uses only DimensionsSpec.getDimensionNames(), and when that list is empty it treats every non-timestamp column as a dimension. It never applies DimensionsSpec.useSchemaDiscovery(), includeAllDimensions, or dimensionExclusions, so V2 ingestion can omit dimensions that discovery/include-all should add or index fields that the ingestion spec explicitly excludes.

Suggestion: Resolve dimensions with the same MapInputRowParser.findDimensions semantics as the normal input-format path, including discovery/include-all behavior and exclusions, before constructing MapBasedInputRow.

FrankChen021 · 2026-09-23T13:03:19Z

-        for (FileScanTask task : tasks) {
-          dataFilePaths.add(task.file().location());
+
+      CloseableIterable<FileScanTask> taskIterable = tableScan.planFiles();


[P3] Close the planned scan iterable

Finding: extractFileScanTasksWithSchema obtains the CloseableIterable returned by TableScan.planFiles() and iterates it without closing it. Iceberg wraps scan planning in a close-completion callback that finalizes planning metrics and closes nested manifest iterables; the old path used try-with-resources. Normal scans therefore skip completion, and exception or early-failure paths can retain planning resources across repeated ingestion attempts.

Suggestion: Iterate the planned tasks inside try-with-resources and close the CloseableIterable on both normal completion and failure.

a2l007 · 2026-09-25T20:43:09Z

@FrankChen021 Please review

FrankChen021

🟡 Changes recommended

The current V2 delete-aware path still has a blocking input-format propagation regression for parallel ingestion, plus a resource-usage issue in the unsplit reader and a failing binaryAsString regression test. Parallel workers can silently diverge from the configured single-task semantics, and large single-task reads can hold too many Parquet file handles open.

Reviewed 10 of 10 changed files.

Scope: updated-since-review incremental diff from 5067eb3, the full merge-base diff, current surrounding Druid input-source/task wiring, and relevant Iceberg/Parquet reader sources.

Validation: static review only; git diff --check passed. No builds, tests, dependency installs, or formatters were run.

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

This is an automated review by Codex GPT-6 Luna(Max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

FrankChen021 · 2026-09-26T10:17:27Z

+  public boolean needsFormat()
+  {
+    // Handles its own reading through Iceberg's native Parquet/ORC reader
+    return false;


[P1] Preserve the configured input format on V2 split workers

Finding: A parallel-index V2 split is deserialized as this source, and Druid passes an InputFormat to a task only when needsFormat() is true. Returning false therefore makes IcebergNativeRecordReader receive null on every worker, so binaryAsString is silently disabled and flattenSpec fields are silently ignored even though the source-level reader and documentation claim to honor or reject those settings. This makes parallel ingestion produce different dimensions and binary values from single-task ingestion.

Suggestion: Carry the relevant input-format settings in the split source or make the split source participate in Druid's input-format propagation so the native reader receives the configured format on worker tasks.

FrankChen021 · 2026-09-26T10:17:27Z

+          public CloseableIterator<InputRow> read(InputStats inputStats) throws IOException
+          {
+            List<CloseableIterator<InputRow>> iters = new ArrayList<>();
+            for (IcebergNativeRecordReader r : taskReaders) {


[P2] Open V2 readers lazily instead of all at once

Finding: The unsplit V2 reader calls read() for every task before returning the concatenated iterator. IcebergNativeRecordReader.read() opens its Parquet iterator during streamRows(), so a table with many data files holds one file handle per data file (and can leak already-open iterators if a later reader fails) until the outer iterator is closed. This can exhaust worker file descriptors for ordinary single-task ingestion.

Suggestion: Build the concatenated iterator lazily and close each task reader before opening the next one, with cleanup for readers opened before an exception.

FrankChen021 · 2026-09-26T10:17:27Z

+   * {@code getBinaryAsString()} accessor, mirroring {@code ParquetInputFormat}'s shape without
+   * depending on the parquet-extensions module.
+   */
+  private static class FakeNestedInputFormat extends NestedInputFormat


[P2] Make the reflective test input format accessible

Finding: The binaryAsString regression test passes a private FakeNestedInputFormat class to IcebergNativeRecordReader, which invokes getBinaryAsString reflectively from a different top-level class. Method.invoke performs access checks on the declaring class, so this private test double causes IllegalAccessException (wrapped as RE) instead of returning the configured value and makes testInputSourceV2HonorsBinaryAsString fail.

Suggestion: Use a public test double or avoid reflective access by exposing a shared accessor contract that the reader can call directly.

FrankChen021

🟡 Changes recommended

The current V2 native reader still mishandles Iceberg temporal values and can silently ignore a valid non-discovering flattenSpec. Tables using native timestamp fields can be ingested with incorrect event times, while flattening behavior can diverge from the V1 path; fix these before merging. The three prior findings on input-format propagation, reader resource usage, and binaryAsString reflection are resolved in the current head.

Reviewed 10 of 10 changed files.

Scope: Started from the incremental diff since 2ad1fca, then reviewed the full merge-base diff and relevant current Druid/Iceberg surrounding code.

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

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

FrankChen021 · 2026-11-02T12:45:22Z

+    if (value == null) {
+      return null;
+    }
+    if (type instanceof Types.TimestampType) {


[P1] Iceberg timestamps stay in microseconds

Finding: GenericParquetReaders returns Iceberg DATE/TIME/TIMESTAMP values in numeric day or microsecond units. This branch only converts Java date-time objects and returns numeric values unchanged, so a real TimestampType used by timestampSpec reaches Druid as microseconds while Druid's numeric timestamp parsing treats it as milliseconds. V2 ingestion therefore assigns event times roughly 1,000 times too large, and date fields have the analogous day-unit mismatch.

Suggestion: Convert Iceberg temporal values by type to the millisecond representation used by the existing Parquet and Arrow readers, and add V2 coverage for timestamp and date columns.

FrankChen021 · 2026-11-02T12:45:22Z

+    }
+    if (inputFormat instanceof NestedInputFormat) {
+      JSONPathSpec flattenSpec = ((NestedInputFormat) inputFormat).getFlattenSpec();
+      if (flattenSpec != null && !flattenSpec.getFields().isEmpty()) {


[P2] Reject unsupported empty flatten specs

Finding: A valid NestedInputFormat can carry JSONPathSpec with useFieldDiscovery=false and an empty fields list. Druid's ObjectFlatteners treats that as an empty flattened row, but this guard rejects only non-empty fields, so the native reader ignores the flattenSpec and emits all schema fields. V2 ingestion can consequently accept rows and derive dimensions that the V1 path would omit or fail on.

Suggestion: Reject every non-null flattenSpec configuration the native reader cannot implement, including useFieldDiscovery=false with no fields, or apply the same flattening behavior before row conversion.

a2l007 added 2 commits May 31, 2026 16:48

Changes to handle iceberg delete files

8a4edf9

Merge branch 'master' of github-personal:apache/druid into iceberg-v2

b2914c8

a2l007 added Area - Batch Ingestion Area - Extension Apache Iceberg https://iceberg.apache.org/ labels Jun 1, 2026

github-actions Bot added the Area - Dependencies label Jun 1, 2026

FrankChen021 reviewed Jun 1, 2026

View reviewed changes

jtuglu1 mentioned this pull request Jun 1, 2026

feat: Iceberg V2 delete file support in druid-iceberg-extensions #19266

Open

6 of 10 tasks

a2l007 added 2 commits September 19, 2026 18:03

Fix warehouseIO

204f5ce

Merge branch 'master' of github-personal:apache/druid into iceberg-v2

42cabdb

github-actions Bot added the Area - Documentation label Sep 20, 2026

a2l007 added 2 commits September 19, 2026 19:03

Fix tests

cbafb90

Upgrade parquet version

5fc1408

a2l007 changed the title ~~Support V2 Format for Iceberg Ingestion~~ feat: Support V2 Format for Iceberg Ingestion Sep 20, 2026

github-advanced-security Bot found potential problems Sep 20, 2026

View reviewed changes

FrankChen021 reviewed Sep 20, 2026

View reviewed changes

Address comments

e7f1b96

github-advanced-security Bot found potential problems Sep 21, 2026

View reviewed changes

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

FrankChen021 reviewed Sep 21, 2026

View reviewed changes

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

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

github-advanced-security Bot found potential problems Sep 22, 2026

View reviewed changes

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

List<Integer> fieldIds = fieldIdsStr.isEmpty()

? Collections.emptyList()

: Arrays.stream(fieldIdsStr.split(","))

.map(Integer::parseInt)

FrankChen021 reviewed Sep 23, 2026

View reviewed changes

Address comments regarding inputformat

2ad1fca

FrankChen021 self-requested a review September 26, 2026 09:14

FrankChen021 reviewed Sep 26, 2026

View reviewed changes

a2l007 added 2 commits November 1, 2026 18:51

Merge branch 'master' of github-personal:apache/druid into iceberg-v2

4a0a49e

Address PR comments

bc60695

FrankChen021 reviewed Nov 2, 2026

View reviewed changes

Conversation

a2l007 commented Jun 1, 2026

Description

Release note

Key changed/added classes in this PR

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

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!

a2l007 commented Sep 25, 2026

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

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

🟡 Changes recommended

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants