datalake/coordinator: add partition overrides to reset

In some cases we want to manually choose from where translation
should resume after a coordinator reset. Add per-partition
last_committed overrides and a reset_all_partitions flag.

By default reset_all_partitions is false and the request is a no-op
unless partition_overrides are provided — only the listed partitions
have their pending entries cleared and last_committed set. Setting
reset_all_partitions to true clears all partitions first, then
applies any overrides. This default-safe design minimizes surprise
and makes it harder to accidentally wipe state.

Author: nvartolomei

proto/redpanda/core/admin/internal/datalake/v1/datalake.proto

@@ -51,10 +51,19 @@ message GetCoordinatorStateResponse {
     CoordinatorState state = 1;
 }
 
+message PartitionStateOverride {
+    optional int64 last_committed = 1;
+}
+
+// When reset_all_partitions is true, clears pending entries and
+// last_committed for all partitions, then applies overrides. When false,
+// clears pending entries only for partitions in partition_overrides and
+// applies their overrides. No-op when false and partition_overrides is empty.
 message CoordinatorResetTopicStateRequest {
     string topic_name = 1;
     int64 revision = 2;
     bool reset_all_partitions = 3;
+    map<int32, PartitionStateOverride> partition_overrides = 4;
 }
 
 message CoordinatorResetTopicStateResponse {}

src/v/datalake/coordinator/BUILD

@@ -51,11 +51,13 @@ redpanda_cc_library(
     ],
     deps = [
         ":file_committer",
+        ":partition_state_override",
         ":snapshot_remover",
         ":state_update",
         ":stm",
         "//src/v/cluster",
         "//src/v/config",
+        "//src/v/container:chunked_hash_map",
         "//src/v/container:chunked_vector",
         "//src/v/model",
         "@abseil-cpp//absl/hash",
@@ -253,6 +255,21 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "partition_state_override",
+    srcs = [
+        "partition_state_override.cc",
+    ],
+    hdrs = [
+        "partition_state_override.h",
+    ],
+    deps = [
+        "//src/v/model",
+        "//src/v/serde",
+        "@fmt",
+    ],
+)
+
 redpanda_cc_library(
     name = "state",
     srcs = [
@@ -281,6 +298,7 @@ redpanda_cc_library(
     ],
     visibility = ["//visibility:public"],
     deps = [
+        ":partition_state_override",
         ":translated_offset_range",
         "//src/v/datalake:schema_identifier",
         "//src/v/datalake:types",
@@ -333,9 +351,11 @@ redpanda_cc_library(
         "state_update.h",
     ],
     deps = [
+        ":partition_state_override",
         ":state",
         ":translated_offset_range",
         "//src/v/base",
+        "//src/v/container:chunked_hash_map",
         "//src/v/container:chunked_vector",
         "//src/v/iceberg:manifest_entry",
         "//src/v/model",

src/v/datalake/coordinator/coordinator.cc

@@ -889,7 +889,9 @@ ss::future<checked<void, coordinator::errc>>
 coordinator::sync_reset_topic_state(
   model::topic topic,
   model::revision_id topic_revision,
-  bool reset_all_partitions) {
+  bool reset_all_partitions,
+  chunked_hash_map<model::partition_id, partition_state_override>
+    partition_overrides) {
     auto gate = maybe_gate();
     if (gate.has_error()) {
         co_return gate.error();
@@ -905,6 +907,7 @@ coordinator::sync_reset_topic_state(
       .topic = topic,
       .topic_revision = topic_revision,
       .reset_all_partitions = reset_all_partitions,
+      .partition_overrides = std::move(partition_overrides),
     };
     auto check_res = update.can_apply(stm_->state());
     if (check_res.has_error()) {

src/v/datalake/coordinator/coordinator.h

@@ -12,8 +12,10 @@
 #include "absl/hash/hash.h"
 #include "cluster/fwd.h"
 #include "config/property.h"
+#include "container/chunked_hash_map.h"
 #include "container/chunked_vector.h"
 #include "datalake/coordinator/file_committer.h"
+#include "datalake/coordinator/partition_state_override.h"
 #include "datalake/coordinator/snapshot_remover.h"
 #include "datalake/coordinator/state_machine.h"
 #include "datalake/fwd.h"
@@ -92,7 +94,9 @@ class coordinator {
     ss::future<checked<void, errc>> sync_reset_topic_state(
       model::topic topic,
       model::revision_id topic_rev,
-      bool reset_all_partitions);
+      bool reset_all_partitions,
+      chunked_hash_map<model::partition_id, partition_state_override>
+        partition_overrides);
 
     void notify_leadership(std::optional<model::node_id>);
 

src/v/datalake/coordinator/frontend.cc

@@ -517,7 +517,8 @@ ss::future<reset_topic_state_reply> frontend::reset_topic_state_locally(
             ->sync_reset_topic_state(
               request.topic,
               request.topic_revision,
-              request.reset_all_partitions)
+              request.reset_all_partitions,
+              std::move(request.partition_overrides))
             .then([](auto result) {
                 reset_topic_state_reply resp{};
                 if (result.has_error()) {

src/v/datalake/coordinator/partition_state_override.cc

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#include "datalake/coordinator/partition_state_override.h"
+
+#include <fmt/ostream.h>
+
+namespace datalake::coordinator {
+
+std::ostream& operator<<(std::ostream& o, const partition_state_override& p) {
+    if (p.last_committed.has_value()) {
+        fmt::print(o, "{{last_committed: {}}}", p.last_committed.value());
+    } else {
+        fmt::print(o, "{{last_committed: nullopt}}");
+    }
+    return o;
+}
+
+} // namespace datalake::coordinator

src/v/datalake/coordinator/partition_state_override.h

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#pragma once
+
+#include "model/fundamental.h"
+#include "serde/envelope.h"
+
+namespace datalake::coordinator {
+
+struct partition_state_override
+  : public serde::envelope<
+      partition_state_override,
+      serde::version<0>,
+      serde::compat_version<0>> {
+    std::optional<kafka::offset> last_committed;
+
+    auto serde_fields() { return std::tie(last_committed); }
+
+    friend std::ostream&
+    operator<<(std::ostream&, const partition_state_override&);
+};
+
+} // namespace datalake::coordinator

src/v/datalake/coordinator/state_update.cc

@@ -324,6 +324,13 @@ reset_topic_state_update::apply(topics_state& state) {
     if (reset_all_partitions) {
         t_state.pid_to_pending_files.clear();
     }
+    for (auto& [pid, po] : partition_overrides) {
+        auto& ps = t_state.pid_to_pending_files[pid];
+        ps.pending_entries.clear();
+        if (po.last_committed.has_value()) {
+            ps.last_committed = po.last_committed.value();
+        }
+    }
     return outcome::success();
 }
 
@@ -368,10 +375,12 @@ std::ostream& operator<<(std::ostream& o, const topic_lifecycle_update& u) {
 std::ostream& operator<<(std::ostream& o, const reset_topic_state_update& u) {
     fmt::print(
       o,
-      "{{topic: {}, revision: {}, reset_all_partitions: {}}}",
+      "{{topic: {}, revision: {}, reset_all_partitions: {}, "
+      "partition_overrides: {} entries}}",
       u.topic,
       u.topic_revision,
-      u.reset_all_partitions);
+      u.reset_all_partitions,
+      u.partition_overrides.size());
     return o;
 }
 

src/v/datalake/coordinator/state_update.h

@@ -10,7 +10,9 @@
 #pragma once
 
 #include "base/outcome.h"
+#include "container/chunked_hash_map.h"
 #include "container/chunked_vector.h"
+#include "datalake/coordinator/partition_state_override.h"
 #include "datalake/coordinator/state.h"
 #include "datalake/coordinator/translated_offset_range.h"
 #include "iceberg/manifest_entry.h"
@@ -127,9 +129,12 @@ struct reset_topic_state_update
     model::topic topic;
     model::revision_id topic_revision;
     bool reset_all_partitions{false};
+    chunked_hash_map<model::partition_id, partition_state_override>
+      partition_overrides;
 
     auto serde_fields() {
-        return std::tie(topic, topic_revision, reset_all_partitions);
+        return std::tie(
+          topic, topic_revision, reset_all_partitions, partition_overrides);
     }
 
     checked<void, stm_update_error> can_apply(const topics_state&);

src/v/datalake/coordinator/tests/BUILD

@@ -122,7 +122,9 @@ redpanda_cc_gtest(
     ],
     deps = [
         ":state_test_utils",
+        "//src/v/container:chunked_hash_map",
         "//src/v/container:chunked_vector",
+        "//src/v/datalake/coordinator:partition_state_override",
         "//src/v/datalake/coordinator:state",
         "//src/v/datalake/coordinator:state_update",
         "//src/v/datalake/coordinator:translated_offset_range",