iceberg: bring back in-manifest stats by andrwng · Pull Request #31332 · redpanda-data/redpanda

andrwng · 2026-07-29T05:00:52Z

Reverts the revert and adds a few more fixes to make it slightly safer in certain pathological cases:

In the coordinator, start counting the iobufs in the stats in the coordinator and apply backpressure based on this metric, similar to how we apply backpressure based on the number of files. The default is to apply a soft limit of 32MiB.
Similarly, when we select groups of files to commit, only collect up to a certain memory (default is 32MiB), similar to how we commit in chunks.
On the translator, per parquet writer, start accounting for the idle memory of each column: previously we were using a fixed value per writer, but that isn't accurate when considering schemas with many fields.
Similarly, add the heap memory of the stats collector to the memory usage accounting.

Backports Required

Release Notes

None

Copilot

Pull request overview

This PR extends the datalake → Iceberg pipeline to propagate Parquet per-column statistics (bounds/counts/sizes) into Iceberg manifest entries, and introduces byte-based bounding/backpressure to keep coordinator/commit memory usage under control.

Changes:

Extract aggregated file-level Parquet column statistics and plumb them through local_file_metadata → coordinator data_file → Iceberg manifest metrics.
Add byte-based caps for commit chunking and coordinator pending backlog (new config tunables + derived-state accounting in coordinator STM).
Add/extend unit + e2e tests covering parquet stats correctness, manifest serialization, and snapshot/backpressure behavior.

Reviewed changes

Copilot reviewed 36 out of 36 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/rptest/tests/datalake/datalake_e2e_test.py	Adds Spark-based e2e assertion that manifest `.files` readable_metrics are populated correctly.
src/v/serde/parquet/writer.h	Exposes file-level aggregated column stats from Parquet writer.
src/v/serde/parquet/writer.cc	Accumulates file-level value counts/sizes and returns per-column file stats.
src/v/serde/parquet/column_writer.h	Adds API to return aggregated per-column stats across row groups.
src/v/serde/parquet/column_writer.cc	Implements file-level stats aggregation/merge across flushes/row groups.
src/v/serde/parquet/BUILD	Makes parquet metadata library publicly visible for downstream use.
src/v/datalake/translation_task.cc	Plumbs local parquet column stats into coordinator `data_file` upload metadata.
src/v/datalake/tests/test_data_writer.h	Fixes test writer to move results out (avoid copies).
src/v/datalake/tests/serde_parquet_writer_test.cc	Adds unit tests validating stats encoding and cross-row-group aggregation.
src/v/datalake/tests/BUILD	Adds deps needed by new parquet stats unit tests.
src/v/datalake/serde_parquet_writer.h	Stores/returns extracted per-column stats post-finish.
src/v/datalake/serde_parquet_writer.cc	Extracts Parquet footer stats into `per_column_stats` for datalake usage.
src/v/datalake/local_parquet_file_writer.cc	Stores per-column stats into `local_file_metadata` after finish.
src/v/datalake/data_writer_interface.h	Extends parquet writer interface with optional `column_stats()` API.
src/v/datalake/coordinator/translated_offset_range.h	Adds estimated-memory helper for pending range accounting.
src/v/datalake/coordinator/tests/state_update_test.cc	Updates tests for new `copy_bounded(max_files, max_bytes, ...)` signature + adds pending totals test.
src/v/datalake/coordinator/tests/state_machine_test.cc	Adds snapshot hydration test ensuring derived pending totals are recomputed.
src/v/datalake/coordinator/tests/iceberg_file_committer_test.cc	Adds tests for manifest column stats + byte-based commit chunking; updates committer ctor usage.
src/v/datalake/coordinator/tests/data_file_test.cc	New tests for estimated-memory sizing behavior with/without column stats.
src/v/datalake/coordinator/tests/coordinator_test.cc	Adds coordinator backpressure test driven by byte limit.
src/v/datalake/coordinator/tests/BUILD	Adds new gtest target and dependencies for added coordinator tests.
src/v/datalake/coordinator/state.h	Adds byte-based bounded copy + derived pending totals API.
src/v/datalake/coordinator/state.cc	Implements byte-bounded copy + running totals + recompute after snapshot install.
src/v/datalake/coordinator/state_update.cc	Maintains running pending totals on add/commit/purge/reset updates.
src/v/datalake/coordinator/state_machine.cc	Recomputes derived pending totals when applying snapshots.
src/v/datalake/coordinator/iceberg_file_committer.h	Adds `max_bytes_per_commit` binding to committer.
src/v/datalake/coordinator/iceberg_file_committer.cc	Writes column metrics into Iceberg manifests + uses byte-bounded `copy_bounded`.
src/v/datalake/coordinator/data_file.h	Adds serialized `column_stats` to coordinator `data_file` + memory estimate helper.
src/v/datalake/coordinator/coordinator.h	Adds byte-based pending backlog limit binding to coordinator.
src/v/datalake/coordinator/coordinator.cc	Uses derived pending totals (files + bytes) for backpressure decision.
src/v/datalake/coordinator/coordinator_manager.cc	Wires new config bindings into coordinator and committer constructors.
src/v/datalake/coordinator/BUILD	Updates deps for coordinator build targets due to new types/usage.
src/v/datalake/BUILD	Adds base_types + parquet metadata deps for parquet stats plumbing.
src/v/datalake/base_types.h	Introduces `per_column_stats` + adds column stats to `local_file_metadata`.
src/v/config/configuration.h	Declares new tunables for byte-based commit/pending caps.
src/v/config/configuration.cc	Defines defaults/help strings for new datalake byte-based tunables.

    _buffered_bytes = _flushed_bytes = 0;
    co_return writer_error::ok;
 }


+        if (cs.bounds.min) {
+            ps.lower_bound = iobuf_to_bytes(cs.bounds.min->value.copy());
+        }
+        if (cs.bounds.max) {
+            ps.upper_bound = iobuf_to_bytes(cs.bounds.max->value.copy());
+        }


        coordinator::data_file uploaded{
          .remote_path = r.value()().string(),
          .row_count = file.local_file.row_count,
          .file_size_bytes = file.local_file.size_bytes,
          .table_schema_id = file.schema_id,
          .partition_spec_id = file.partition_spec_id,
          .partition_key = std::move(pk_fields),
+          .column_stats = file.local_file.column_stats.copy(),
        };


vbotbuildovich · 2026-07-29T06:50:47Z

CI test results

test results on build#87868

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FLAKY(PASS)	ShadowLinkTopicFailoverTests	test_producer_ids_failover	{"storage_mode": "tiered_v1"}	integration	https://buildkite.com/redpanda/redpanda/builds/87868#019fac5a-30b7-43ec-8eb4-28b553a4c609	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0114, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=ShadowLinkTopicFailoverTests&test_method=test_producer_ids_failover
FLAKY(PASS)	ShadowLinkingMetricsTests	test_link_metrics	null	integration	https://buildkite.com/redpanda/redpanda/builds/87868#019fac5a-30ba-40c1-b277-61843cfb071f	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0031, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=ShadowLinkingMetricsTests&test_method=test_link_metrics
FLAKY(PASS)	NodeWiseRecoveryTest	test_recovery_local_data_missing	{"wait_for_final_manifest_uploads": true}	integration	https://buildkite.com/redpanda/redpanda/builds/87868#019fac5a-30b7-43ec-8eb4-28b553a4c609	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0182, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=NodeWiseRecoveryTest&test_method=test_recovery_local_data_missing
FLAKY(PASS)	SimpleEndToEndTest	test_relaxed_acks	{"write_caching": false}	integration	https://buildkite.com/redpanda/redpanda/builds/87868#019fac5c-8104-4901-9013-8e6f9aa3a686	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0054, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=SimpleEndToEndTest&test_method=test_relaxed_acks

test results on build#87960

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FAIL	ParquetWriter	ColumnMemoryEstimateCoversActual		unit	https://buildkite.com/redpanda/redpanda/builds/87960#019fb0ea-720c-428e-91de-83881b9c2197	0/1

test results on build#88390

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FAIL	MasterTestSuite	per_entity_probe_deregistered_on_gc		unit	https://buildkite.com/redpanda/redpanda/builds/88390#019fd934-903a-4f6b-a5d2-378816796659	0/1

andrwng · 2026-07-30T18:49:26Z

/ci-repeat 1
dt-repeat=10
skip-redpanda-build
skip-units
skip-rebase
tests/rptest/tests/datalake/custom_partitioning_test.py::DatalakeCustomPartitioningTest.test_many_partitions@{"catalog_type":"rest_jdbc","cloud_storage_type":1}

andrwng · 2026-07-30T19:45:55Z

/ci-repeat 10
dt-repeat=20
skip-redpanda-build
skip-units
skip-rebase
tests/rptest/tests/datalake/custom_partitioning_test.py::DatalakeCustomPartitioningTest.test_many_partitions@{"catalog_type":"rest_jdbc","cloud_storage_type":1}

wdberkeley

Nice. This is a big improvement.

wdberkeley · 2026-08-03T18:56:06Z

+      state.pending_files() >= max_pending_files_()
+      || state.pending_bytes() >= max_pending_bytes_()) {
+        backpressured_as_of_ = now;
+        return true;


has_too_many_pending_files is technically a misnomer as we may have few files but their pending state takes too much memory. If you can come up with a good name it's worth renaming, but it's also ~fitting.

Also is it worth distinguishing (for an error message, metric) the cause of the backpressure, returning an enum or error code instead of a boolean?

Only did a rename, opted out of an enum/error code. Seemed like kind of overkill but can add it if you feel strongly about it

Don't feel strongly. Fine as is.

wdberkeley · 2026-08-03T19:06:25Z

+// The estimate models seastar-allocator resident bytes, which is what the
+// translator's memory semaphore admits writers against, so the measurement is
+// only meaningful in builds that link that allocator.
+TEST(ParquetWriter, ColumnMemoryEstimateCoversActual) {


This test is very Claude. It's a little sus but if it doesn't flake or act weird it's nice to have.

Yea kinda agree, but it's pretty consistent it seems. And it's useful for picking a reasonable estimate (I think)

wdberkeley · 2026-08-03T19:08:27Z

+    // run that test (it prints the measured per-leaf on failure) and set this
+    // so the estimate falls back in range. Absolute bytes are allocator
+    // dependent, so the test compares a large-N slope, not an absolute figure.
+    constexpr size_t largest_column_object = std::max({


Funny to see the AI go to such lengths with the test and the comment to guard the code against drift of the estimate, but this list would silently go stale if we add a physical type. 🤷‍♂️

Yea, fair. The whole memory estimation is empirical anyway so I removed this.

Thread per-column stats (min/max bounds, null counts, value counts, column sizes) through to the Iceberg data_file manifest entry, where query engines use them for column-level predicate pushdown. Maintain a file-level column_stats_collector in buffered_column_writer that accumulates by merging after each flush_pages(), then use the result after the file is done to get file-level stats. (cherry picked from commit 7f83353)

(cherry picked from commit b4e8232)

Adds estimated_memory_bytes() for data_file and translated_offset_range. For use by a subsequent commit that bounds the coordinator's pending state by memory rather than file count.

has_too_many_pending_files() scans every pending entry on each request. Rather than keep scanning as we start to also account memory, maintain running totals on topics_state, updated as entries enter and leave pending state. The totals are derived, so they're left out of serde_fields() and rebuilt by recompute_pending() on snapshot install. Underflow is clamped rather than left to wrap: a wrapped total would sit above every backpressure limit and reject adds for the whole coordinator until the next recompute. For use by subsequent commits that shed load and chunk commits by memory.

A pending file's footprint varies widely with its column stats, so a file count is a poor proxy for the memory has_too_many_pending_files() is meant to bound. Adds datalake_coordinator_max_pending_bytes (soft, 32 MiB) and sheds when either total is over its limit. Reads the running totals, so the check is O(1) rather than a scan. The ducktape case puts the file count out of reach so nothing but the byte limit can reject a request, which makes the backpressure metrics on their own proof that the path works.

copy_bounded() caps the files copied into a single commit pass to bound its transient memory, but the same file count can hold far more memory once files carry column stats. Adds datalake_coordinator_max_bytes_per_commit (soft, 32 MiB) and stops the copy once either limit is hit.

Adds writer::estimated_memory(num_leaf_columns): sizeof(impl) plus a per-column term. That term is empirical, covering the column writer object, the chunks its containers take on the first row, the schema element and allocator rounding. ColumnMemoryEstimateCoversActual reports what has to be covered: estimated_column_memory() needs at least 2920 to cover a byte_array column; it is 4096 The slope holds near 2900 whether the second measurement is 2000 or 10000 columns, so 4KiB leaves about 40% over it. It only measures where seastar's allocator is linked, and skips elsewhere. For use by the datalake writer memory reservation in a subsequent commit.

Each open parquet writer costs the output stream buffer, held until close, plus a column writer per leaf column. A high-cardinality partition spec opens a writer per partition value, which can be far more than the datalake memory pool holds. The reservation is released on flush, so writers that stay open past one still hold memory the pool no longer accounts for.

column_writer::memory_usage() drives the translator's memory reservation but counted only buffered page data. Each leaf column also retains the row group's min/max bounds untruncated, which for byte-array columns grows with the value width.

iobuf_to_bytes() takes its argument by const ref and copies into a fresh bytes, so copying the bound first was an extra allocation and memcpy per column on every file finish.

andrwng · 2026-08-06T22:33:45Z

Latest push was a rebase.

Diff from right before then since the last review: https://github.com/redpanda-data/redpanda/compare/9c0c7c65f98846843c75c33f1b28eda526262a4f..4fbc54521534870c4d276a655e36f9320898871c

vbotbuildovich · 2026-08-07T19:04:09Z

/backport v26.2.x

vbotbuildovich · 2026-08-07T19:04:10Z

/backport v26.1.x

vbotbuildovich · 2026-08-07T19:04:10Z

/backport v25.3.x

vbotbuildovich · 2026-08-07T19:14:22Z

Failed to create a backport PR to v25.3.x branch. I tried:

git remote add upstream https://github.com/redpanda-data/redpanda.git
git fetch --all
git checkout -b backport-pr-31332-v25.3.x-70 remotes/upstream/v25.3.x
git cherry-pick -x ddb4ace072 fcad9e1362 f3cfe07a0e 9a98ac5ee1 e12338af5b cb7ebf5905 5497e8fc15 3eaf8279bc babc950f87 5e64e28bf6

Workflow run logs.

vbotbuildovich · 2026-08-07T19:16:18Z

Failed to create a backport PR to v26.1.x branch. I tried:

git remote add upstream https://github.com/redpanda-data/redpanda.git
git fetch --all
git checkout -b backport-pr-31332-v26.1.x-755 remotes/upstream/v26.1.x
git cherry-pick -x ddb4ace072 fcad9e1362 f3cfe07a0e 9a98ac5ee1 e12338af5b cb7ebf5905 5497e8fc15 3eaf8279bc babc950f87 5e64e28bf6

Workflow run logs.

Copilot AI review requested due to automatic review settings July 29, 2026 05:00

andrwng requested a review from a team as a code owner July 29, 2026 05:00

github-actions Bot added area/build area/redpanda labels Jul 29, 2026

Copilot started reviewing on behalf of andrwng July 29, 2026 05:01 View session

Copilot AI reviewed Jul 29, 2026

View reviewed changes

andrwng force-pushed the iceberg-manifest-stats branch from 1c96301 to 9c0c7c6 Compare July 30, 2026 08:51

andrwng changed the title ~~wip: iceberg manifest stats redux~~ iceberg: bring back in-manifest stats Jul 31, 2026

wdberkeley self-requested a review August 3, 2026 15:52

wdberkeley previously approved these changes Aug 3, 2026

View reviewed changes

andrwng dismissed wdberkeley’s stale review via 09aa654 August 5, 2026 07:10

andrwng force-pushed the iceberg-manifest-stats branch from 9c0c7c6 to 09aa654 Compare August 5, 2026 07:10

andrwng requested a review from wdberkeley August 5, 2026 07:18

andrwng force-pushed the iceberg-manifest-stats branch 2 times, most recently from 29b9ad4 to 4fbc545 Compare August 6, 2026 22:13

wdberkeley and others added 10 commits August 6, 2026 15:16

rptest/datalake: test Iceberg column stats in manifest

fcad9e1

(cherry picked from commit b4e8232)

dl/coordinator: estimate in-memory footprint of pending files

f3cfe07

Adds estimated_memory_bytes() for data_file and translated_offset_range. For use by a subsequent commit that bounds the coordinator's pending state by memory rather than file count.

datalake: remove redundant copy of stats bounds

5e64e28

iobuf_to_bytes() takes its argument by const ref and copies into a fresh bytes, so copying the bound first was an extra allocation and memcpy per column on every file finish.

andrwng force-pushed the iceberg-manifest-stats branch from 4fbc545 to 5e64e28 Compare August 6, 2026 22:31

wdberkeley approved these changes Aug 7, 2026

View reviewed changes

andrwng merged commit ad75461 into redpanda-data:dev Aug 7, 2026
19 checks passed

vbotbuildovich mentioned this pull request Aug 7, 2026

[v26.2.x] iceberg: bring back in-manifest stats #31496

Merged

vbotbuildovich mentioned this pull request Aug 7, 2026

[v25.3.x] iceberg: bring back in-manifest stats #31497

Open

vbotbuildovich mentioned this pull request Aug 7, 2026

[v26.1.x] iceberg: bring back in-manifest stats #31498

Open

Uh oh!

Conversation

andrwng commented Jul 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Backports Required

Release Notes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

vbotbuildovich commented Jul 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI test results

Uh oh!

andrwng commented Jul 30, 2026

Uh oh!

andrwng commented Jul 30, 2026

Uh oh!

wdberkeley left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

andrwng commented Aug 6, 2026

Uh oh!

Uh oh!

vbotbuildovich commented Aug 7, 2026

Uh oh!

vbotbuildovich commented Aug 7, 2026

Uh oh!

vbotbuildovich commented Aug 7, 2026

Uh oh!

vbotbuildovich commented Aug 7, 2026

Uh oh!

vbotbuildovich commented Aug 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

andrwng commented Jul 29, 2026 •

edited

Loading

vbotbuildovich commented Jul 29, 2026 •

edited

Loading