Iceberg backlog controller #24990

mmaslankaprv · 2025-01-31T14:53:39Z

Added datalake backlog controller to dynamically adjust the datalake
scheduling group shares. Simple proportional controller tracks the
average size of partition translation backlog and calculates how far is
it from the desired value. The number of shares given to the datalake
scheduling group is proportional to the difference between the desired
and current backlog values.

The rationale behind the controller is that datalake translation is
almost always CPU bound and the translation rate (bytes of translated
bytes per second) can be controlled by giving the translator more CPU
time when required.

Backports Required

Release Notes

Improvements

controlling Iceberg translation backlog

Previously the `disk_log_impl::size_bytes_after_offset` was returning a very rough approximation of the partition size after offset as it limited the result accuracy to the segment sized. Changed the code to return a better approximation of size after offset by using a segment index. When requested offset is within the segment a closest segment index entry is queried, the entry contains a position in segment file, the position is an offset in bytes from the beginning of the file. By subtracting the position from file size the code calculates the remaining segment part with the accuracy of segment index entry interval which by default is equal to 32KiB. Signed-off-by: Michał Maślanka <[email protected]>

Added `partition_translator` method that calculates the translation backlog. A backlog is a number of bytes in partition that are ready to be translated but did not yet went through the translation process. Signed-off-by: Michał Maślanka <[email protected]>

Added properties that are responsible for changing behavior of backlog controller. Signed-off-by: Michał Maślanka <[email protected]>

Added datalake backlog controller to dynamically adjust the `datalake` scheduling group shares. Simple proportional controller tracks the average size of partition translation backlog and calculates how far is it from the desired value. The number of shares given to the `datalake` scheduling group is proportional to the difference between the desired and current backlog values. The rationale behind the controller is that datalake translation is almost always CPU bound and the translation rate (bytes of translated bytes per second) can be controlled by giving the translator more CPU time when required. Signed-off-by: Michał Maślanka <[email protected]>

vbotbuildovich · 2025-01-31T19:37:04Z

CI test results

test results on build#61441

test_id	test_kind	job_url	test_status	passed
rptest.tests.datalake.compaction_test.CompactionGapsTest.test_translation_no_gaps.cloud_storage_type=CloudStorageType.S3	ducktape	https://buildkite.com/redpanda/redpanda/builds/61441#0194bd63-58b6-4a65-b64b-e15a9b2bc4d5	FLAKY	1/3

andrwng

This is cool! Have you tried it out on a cluster yet? Curious if you have any empirical improvements worth sharing

andrwng · 2025-01-31T18:30:46Z

src/v/storage/disk_log_impl.cc

@@ -1845,8 +1845,21 @@ uint64_t disk_log_impl::size_bytes_after_offset(model::offset o) const {
    uint64_t size = 0;
    for (size_t i = _segs.size(); i-- > 0;) {
        auto& seg = _segs[i];
-        if (seg->offsets().get_base_offset() < o) {
-            break;
+        auto& offsets = seg->offsets();


Just curious if you think this is worth backporting? Seems this function is used in Raft at least and I'm wondering the effects on other callers

andrwng · 2025-01-31T21:01:32Z

src/v/datalake/translation/partition_translator.cc

@@ -340,6 +340,32 @@ partition_translator::do_translation_for_range(
    co_return std::move(result.value());
 }

+std::optional<size_t> partition_translator::translation_backlog() const {


I'm curious if there's actually correlation between translation time and bytes on disk, particularly in the face of compressed batches, or complex schemas. I guess the thinking is larger messages likely imply more complex payload, but just thinking out loud I wonder if something as simple as number of records pending would be usable here.

Not really advocating for any changes, but just thought I'd pose the question, and also point out that this won't work for read replicas or restored/FPM-ed partitions without implementing a similar sizing method for cloud.

andrwng · 2025-01-31T21:04:27Z

src/v/config/configuration.cc

+      "Average size of the datalake translation backlog, per partition, that "
+      "the "
+      "backlog controller will try to maintain.",


Just want to make sure I understand, does this mean that when we have more than this value not translated, translation is scheduled more aggressively? If so, maybe worth explicitly writing that

andrwng · 2025-01-31T21:05:46Z

src/v/config/configuration.cc

+      "Proportional coefficient for the Iceberg backlog controller. Number of "
+      "shares assigned to the datalake scheduling group will be proportional "
+      "to the backlog size error.",


I think it's worth leaving a note around tuning this, like if more negative values mean that the system has more aggressive adaptivity to workloads or something

andrwng · 2025-01-31T21:09:45Z

src/v/datalake/backlog_controller.h

+    config::binding<double> _proportional_coeff;
+    config::binding<uint32_t> _setpoint;
+    std::chrono::steady_clock::duration _sampling_interval;
+    ss::timer<> _sampling_timer;
+    long double _current_sample{0};


nit: mind adding some docs about these? In particular i think it's helpful to at least understand:

setpoint is the target value we're trying to maintain

current sample has the same units as setpoint

current sample is an output of _sampling_f

maybe how to think about the coefficient, though maybe that's more useful to put in the cluster config property description?

andrwng · 2025-01-31T21:11:07Z

src/v/datalake/backlog_controller.cc

+      _setpoint(),
+      _current_sample,
+      current_err,
+


nit: extra newline, and update used twice?

bharathv

lgtm generally, few clarifying questions

bharathv · 2025-01-31T20:15:22Z

src/v/datalake/backlog_controller.h

+    ss::timer<> _sampling_timer;
+    long double _current_sample{0};
+
+    int _min_shares{1};


wondering if min should be at least default 100.

bharathv · 2025-01-31T20:17:50Z

src/v/datalake/backlog_controller.cc

+      update,
+      update);
+
+    _scheduling_group.set_shares(static_cast<float>(update));


Is there any use to just adjust the weights of translation and produce, something like

translation_shares + delta
produce_shares - delta

In the current form I think it shuffles the entire share distribution which effectively means (for example) fetch gets fewer shares if there is a translation lag, is that the intention?

bharathv · 2025-01-31T21:11:45Z

src/v/config/configuration.cc

+      *this,
+      "iceberg_target_backlog_size",
+      "Average size of the datalake translation backlog, per partition, that "
+      "the "


nit: formatting

bharathv · 2025-01-31T21:13:06Z

src/v/datalake/datalake_manager.cc

+        total_lag += backlog_size.value();
+    }
+
+    return total_lag / _translators.size();


/ _translators.size();

nit: account for translators that only report something?

bharathv · 2025-02-01T02:22:28Z

src/v/config/configuration.cc

+      "the "
+      "backlog controller will try to maintain.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      5_MiB,


I was initially considering whether the default size should be slightly larger than 5 MiB. However, a smaller default has the advantage of smaller proportional changes to scheduling weights. This allows the system to correct itself more quickly than with a larger default (e.g., 1 GiB), where weight changes would be drastic and throttling more significant (bigger sawtooth). Therefore, we likely prefer smaller values for faster system adjustment, am I thinking about it correctly?

mmaslankaprv requested a review from a team as a code owner January 31, 2025 14:53

mmaslankaprv marked this pull request as draft January 31, 2025 14:53

github-actions bot added area/build area/redpanda labels Jan 31, 2025

mmaslankaprv added 2 commits January 31, 2025 15:54

mmaslankaprv force-pushed the icebreg-backlog-controller branch from f9e0982 to 3f2293e Compare January 31, 2025 14:54

mmaslankaprv added 2 commits January 31, 2025 16:41

config: introduced properties for datalake backlog controller

decde47

Added properties that are responsible for changing behavior of backlog controller. Signed-off-by: Michał Maślanka <[email protected]>

mmaslankaprv force-pushed the icebreg-backlog-controller branch from 3f2293e to 29b8682 Compare January 31, 2025 15:41

mmaslankaprv marked this pull request as ready for review January 31, 2025 15:55

mmaslankaprv requested review from bharathv, bashtanov, ztlpn and andrwng January 31, 2025 16:36

andrwng reviewed Jan 31, 2025

View reviewed changes

bharathv reviewed Feb 1, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Iceberg backlog controller #24990

Iceberg backlog controller #24990

mmaslankaprv commented Jan 31, 2025

vbotbuildovich commented Jan 31, 2025

andrwng left a comment

andrwng Jan 31, 2025

andrwng Jan 31, 2025

andrwng Jan 31, 2025

andrwng Jan 31, 2025

andrwng Jan 31, 2025

andrwng Jan 31, 2025

bharathv left a comment

bharathv Jan 31, 2025

bharathv Jan 31, 2025

bharathv Jan 31, 2025

bharathv Jan 31, 2025

bharathv Feb 1, 2025

Iceberg backlog controller #24990

Are you sure you want to change the base?

Iceberg backlog controller #24990

Conversation

mmaslankaprv commented Jan 31, 2025

Backports Required

Release Notes

Improvements

vbotbuildovich commented Jan 31, 2025

CI test results

andrwng left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bharathv left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment