diff --git a/include/paimon/file_store_write.h b/include/paimon/file_store_write.h
index 8859e4f3f..5e6b253c5 100644
--- a/include/paimon/file_store_write.h
+++ b/include/paimon/file_store_write.h
@@ -21,6 +21,7 @@
 #include <cstdint>
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -145,6 +146,18 @@ class PAIMON_EXPORT FileStoreWrite {
     /// an old context is unsupported.
     virtual Status RefreshCommittedSnapshot(int64_t snapshot_id);
 
+    /// Returns the base snapshot for an existing partition-bucket writer without storage I/O.
+    ///
+    /// Initialized from the bucket's restore snapshot and advanced after a successful real-time
+    /// refresh of this bucket. A failed refresh leaves the bucket's value unchanged.
+    ///
+    /// @param partition Logical partition values.
+    /// @param bucket Bucket id of an existing writer.
+    /// @return The snapshot id, or nullopt if no snapshot has been used. Returns an error for a
+    ///         missing bucket writer or a format table.
+    virtual Result<std::optional<int64_t>> GetBaseSnapshotId(
+        const std::map<std::string, std::string>& partition, int32_t bucket) const = 0;
+
     virtual std::shared_ptr<Metrics> GetMetrics() const = 0;
 
     /// Releases resources owned by this writer.
diff --git a/src/paimon/core/operation/abstract_file_store_write.cpp b/src/paimon/core/operation/abstract_file_store_write.cpp
index a8eecace4..3c8e19114 100644
--- a/src/paimon/core/operation/abstract_file_store_write.cpp
+++ b/src/paimon/core/operation/abstract_file_store_write.cpp
@@ -496,6 +496,42 @@ std::shared_ptr<Metrics> AbstractFileStoreWrite::GetMetrics() const {
     return result;
 }
 
+Result<std::optional<int64_t>> AbstractFileStoreWrite::GetBaseSnapshotId(
+    const std::map<std::string, std::string>& partition, int32_t bucket) const {
+    PAIMON_ASSIGN_OR_RAISE(BinaryRow part, file_store_path_factory_->ToBinaryRow(partition));
+    std::optional<int64_t> snapshot_id;
+    {
+        std::lock_guard<std::mutex> lock(writers_mutex_);
+        auto partition_iter = writers_.find(part);
+        if (partition_iter == writers_.end()) {
+            return Status::KeyError("partition-bucket writer does not exist");
+        }
+        auto bucket_iter = partition_iter->second.find(bucket);
+        if (bucket_iter == partition_iter->second.end()) {
+            return Status::KeyError("partition-bucket writer does not exist");
+        }
+        snapshot_id = bucket_iter->second.base_snapshot_id;
+    }
+    std::shared_ptr<RealtimeContext> context = GetRealtimeContext();
+    if (context) {
+        std::vector<std::pair<std::string, std::string>> partition_values;
+        PAIMON_ASSIGN_OR_RAISE(partition_values,
+                               file_store_path_factory_->GeneratePartitionVector(part));
+        const std::map<std::string, std::string> canonical_partition(partition_values.begin(),
+                                                                     partition_values.end());
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RealtimeContextImpl> context_impl,
+                               RealtimeContextImpl::Cast(context));
+        PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> refreshed_snapshot_id,
+                               context_impl->GetRefreshedSnapshotId(
+                                   RealtimePartitionBucket(canonical_partition, bucket)));
+        // Lazy restore may have selected a newer snapshot than the context's last refresh.
+        if (refreshed_snapshot_id && (!snapshot_id || *snapshot_id < *refreshed_snapshot_id)) {
+            snapshot_id = refreshed_snapshot_id;
+        }
+    }
+    return snapshot_id;
+}
+
 Status AbstractFileStoreWrite::CheckRealtimeWriteUsable() const {
     if (IsRealtimeWrite() && realtime_write_poisoned_.load()) {
         return Status::Invalid(
@@ -623,16 +659,20 @@ Result<std::shared_ptr<BatchWriter>> AbstractFileStoreWrite::GetWriter(const Bin
         std::shared_ptr<BatchWriter> writer,
         CreateWriter(partition, bucket, restore_data_files, max_sequence_number, dv_maintainer));
     int32_t total_buckets = restored->TotalBuckets().value_or(GetDefaultBucketNum());
+    const std::optional<Snapshot> snapshot = restored->GetSnapshot();
+    const std::optional<int64_t> base_snapshot_id =
+        snapshot ? std::optional<int64_t>(snapshot->Id()) : std::nullopt;
 
     if (partition_iter == writers_.end()) {
         writers_.emplace(
-            partition, std::unordered_map<int32_t, WriterContainer<BatchWriter>>(
-                           {{bucket, WriterContainer<BatchWriter>(writer, total_buckets,
-                                                                  primary_key_index_maintainer)}}));
+            partition,
+            std::unordered_map<int32_t, WriterContainer<BatchWriter>>(
+                {{bucket, WriterContainer<BatchWriter>(writer, total_buckets, base_snapshot_id,
+                                                       primary_key_index_maintainer)}}));
     } else {
         partition_iter->second.emplace(
-            bucket,
-            WriterContainer<BatchWriter>(writer, total_buckets, primary_key_index_maintainer));
+            bucket, WriterContainer<BatchWriter>(writer, total_buckets, base_snapshot_id,
+                                                 primary_key_index_maintainer));
     }
     writer_memory_manager_->RegisterWriter(writer.get());
 
diff --git a/src/paimon/core/operation/abstract_file_store_write.h b/src/paimon/core/operation/abstract_file_store_write.h
index d98cc8706..644d88dd7 100644
--- a/src/paimon/core/operation/abstract_file_store_write.h
+++ b/src/paimon/core/operation/abstract_file_store_write.h
@@ -23,6 +23,7 @@
 #include <limits>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <shared_mutex>
 #include <string>
 #include <unordered_map>
@@ -102,6 +103,8 @@ class AbstractFileStoreWrite : public FileStoreWrite {
         int64_t commit_identifier) override;
     Status Close() override;
     std::shared_ptr<Metrics> GetMetrics() const override;
+    Result<std::optional<int64_t>> GetBaseSnapshotId(
+        const std::map<std::string, std::string>& partition, int32_t bucket) const override;
 
     const CoreOptions& GetOptions() const {
         return options_;
@@ -113,13 +116,16 @@ class AbstractFileStoreWrite : public FileStoreWrite {
         WriterContainer() = default;
         WriterContainer(
             const std::shared_ptr<T>& writer, int32_t total_buckets,
+            const std::optional<int64_t>& base_snapshot_id,
             const std::shared_ptr<BucketedPrimaryKeyIndexMaintainer>& primary_key_index_maintainer)
             : writer(writer),
               total_buckets(total_buckets),
+              base_snapshot_id(base_snapshot_id),
               primary_key_index_maintainer(primary_key_index_maintainer) {}
         std::shared_ptr<T> writer;
         int64_t last_modified_commit_identifier = std::numeric_limits<int64_t>::min();
         int32_t total_buckets = -1;
+        std::optional<int64_t> base_snapshot_id;
         std::shared_ptr<BucketedPrimaryKeyIndexMaintainer> primary_key_index_maintainer;
     };
 
@@ -179,7 +185,7 @@ class AbstractFileStoreWrite : public FileStoreWrite {
  private:
     std::unordered_map<BinaryRow, std::unordered_map<int32_t, WriterContainer<BatchWriter>>>
         writers_;
-    std::mutex writers_mutex_;
+    mutable std::mutex writers_mutex_;
     std::mutex realtime_seal_prepare_mutex_;
     // Seal/prepare may flush concurrently with writes, but cannot return before an in-flight
     // write has published its outcome, including invalidating the context on failure.
diff --git a/src/paimon/core/operation/file_store_write_test.cpp b/src/paimon/core/operation/file_store_write_test.cpp
index 520e850e5..01f46b053 100644
--- a/src/paimon/core/operation/file_store_write_test.cpp
+++ b/src/paimon/core/operation/file_store_write_test.cpp
@@ -99,8 +99,11 @@ TEST(FileStoreWriteTest, TestRealtimeCatalogCommitRefreshAndRecovery) {
         };
         ASSERT_OK_AND_ASSIGN(auto realtime_context, RealtimeContext::Create());
         ASSERT_OK_AND_ASSIGN(auto writer, create_writer(realtime_context));
+        ASSERT_NOK_WITH_MSG(writer->GetBaseSnapshotId({}, 0), "writer does not exist");
         ASSERT_OK_AND_ASSIGN(auto batch, make_batch(R"([[10, 1, "a"], [20, 2, "b"]])"));
         ASSERT_OK(writer->Write(std::move(batch)));
+        ASSERT_OK_AND_ASSIGN(auto base_snapshot_id, writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, std::nullopt);
         ASSERT_OK_AND_ASSIGN(auto progress, writer->PrepareCommitWithProgress(1));
         ASSERT_EQ(progress.size(), 1u);
         CommitContextBuilder commit_builder(table_path, "writer");
@@ -117,11 +120,21 @@ TEST(FileStoreWriteTest, TestRealtimeCatalogCommitRefreshAndRecovery) {
                                                     table_path, "snapshot/snapshot-1")));
         ASSERT_FALSE(has_snapshot);
 
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, std::nullopt);
+        catalog->SetLoadSnapshotStatus(Status::IOError("catalog unavailable"));
+        ASSERT_NOK_WITH_MSG(writer->RefreshCommittedSnapshot(snapshot_id), "catalog unavailable");
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, std::nullopt);
+        catalog->SetLoadSnapshotStatus(Status::OK());
         ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot_id));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, 1);
         ASSERT_OK_AND_ASSIGN(double retained_rows,
                              writer->GetMetrics()->GetGauge(RealtimeMetrics::kTotalRowCount));
         ASSERT_EQ(retained_rows, 0);
         ASSERT_OK(writer->Close());
+        ASSERT_NOK_WITH_MSG(writer->GetBaseSnapshotId({}, 0), "writer does not exist");
 
         ASSERT_OK_AND_ASSIGN(auto recovered_context, RealtimeContext::Create());
         ASSERT_OK_AND_ASSIGN(auto recovered_writer, create_writer(recovered_context));
@@ -131,11 +144,23 @@ TEST(FileStoreWriteTest, TestRealtimeCatalogCommitRefreshAndRecovery) {
         ASSERT_EQ(read_state.committed_offsets, expected_offsets);
         ASSERT_OK_AND_ASSIGN(auto next_batch, make_batch(R"([[21, 1, "updated"]])"));
         ASSERT_OK(recovered_writer->Write(std::move(next_batch)));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, recovered_writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, 1);
+        ASSERT_OK_AND_ASSIGN(auto shared_writer, create_writer(recovered_context));
+        ASSERT_OK_AND_ASSIGN(auto empty_batch, make_batch("[]"));
+        ASSERT_OK(shared_writer->Write(std::move(empty_batch)));
         ASSERT_OK_AND_ASSIGN(auto next_progress, recovered_writer->PrepareCommitWithProgress(2));
         ASSERT_OK_AND_ASSIGN(snapshot_id,
                              commit->CommitWithProgress(next_progress, 2, std::nullopt));
         ASSERT_EQ(snapshot_id, 2);
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, recovered_writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, 1);
         ASSERT_OK(recovered_writer->RefreshCommittedSnapshot(snapshot_id));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, recovered_writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, 2);
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, shared_writer->GetBaseSnapshotId({}, 0));
+        ASSERT_EQ(base_snapshot_id, 2);
+        ASSERT_OK(shared_writer->Close());
         ASSERT_OK(recovered_writer->Close());
 
         catalog->SetLoadSnapshotStatus(Status::IOError("catalog unavailable"));
@@ -144,6 +169,110 @@ TEST(FileStoreWriteTest, TestRealtimeCatalogCommitRefreshAndRecovery) {
     }
 }
 
+TEST(FileStoreWriteTest, TestBucketBaseSnapshotTracksActualLazyRestore) {
+    for (bool primary_key : {false, true}) {
+        SCOPED_TRACE(primary_key);
+        auto dir = UniqueTestDirectory::Create();
+        ASSERT_TRUE(dir);
+        const Identifier identifier("db", "table");
+        const auto logical_schema = arrow::schema(
+            {arrow::field("dt", arrow::utf8(), false), arrow::field("id", arrow::int64(), false)});
+        const std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                            {Options::BUCKET_KEY, "id"},
+                                                            {Options::FILE_FORMAT, "parquet"},
+                                                            {Options::WRITE_ONLY, "true"}};
+        const std::vector<std::string> primary_keys =
+            primary_key ? std::vector<std::string>{"dt", "id"} : std::vector<std::string>{};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> schema,
+                             TableSchema::Create(0, logical_schema, {"dt"}, primary_keys, options));
+        auto catalog = std::make_shared<MockVersionManagedCatalog>();
+        catalog->SetTableSchema(schema);
+        catalog->SetFileSystem(dir->GetFileSystem());
+        catalog->CheckBaseSnapshotUuid();
+        auto create_writer = [&](bool ignore_previous) -> Result<std::unique_ptr<FileStoreWrite>> {
+            WriteContextBuilder builder(dir->Str(), "writer");
+            PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> context,
+                                   builder.WithCatalog(catalog, identifier)
+                                       .WithStreamingMode(true)
+                                       .WithIgnorePreviousFiles(ignore_previous)
+                                       .Finish());
+            return FileStoreWrite::Create(std::move(context));
+        };
+        auto make_batch = [&](const std::string& dt, int32_t bucket,
+                              const std::string& json) -> Result<std::unique_ptr<RecordBatch>> {
+            PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+                                              arrow::ipc::internal::json::ArrayFromJSON(
+                                                  arrow::struct_(logical_schema->fields()), json));
+            ArrowArray c_array;
+            PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array));
+            return RecordBatchBuilder(&c_array)
+                .SetPartition({{"dt", dt}})
+                .SetBucket(bucket)
+                .Finish();
+        };
+        CommitContextBuilder commit_builder(dir->Str(), "writer");
+        ASSERT_OK_AND_ASSIGN(auto commit_context,
+                             commit_builder.WithCatalog(catalog, identifier).Finish());
+        ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context)));
+        auto seed_snapshot = [&](int32_t bucket, int64_t identifier) -> Status {
+            PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreWrite> writer, create_writer(false));
+            PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<RecordBatch> batch,
+                                   make_batch("a", bucket, R"([["a", 1]])"));
+            PAIMON_RETURN_NOT_OK(writer->Write(std::move(batch)));
+            PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<CommitMessage>> messages,
+                                   writer->PrepareCommit(false, identifier));
+            PAIMON_RETURN_NOT_OK(commit->Commit(messages, identifier));
+            return writer->Close();
+        };
+
+        ASSERT_OK_AND_ASSIGN(auto writer, create_writer(false));
+        ASSERT_NOK_WITH_MSG(writer->GetBaseSnapshotId({{"dt", "empty"}}, 0),
+                            "writer does not exist");
+        ASSERT_OK_AND_ASSIGN(auto batch, make_batch("empty", 0, R"([["empty", 2]])"));
+        ASSERT_OK(writer->Write(std::move(batch)));
+        ASSERT_OK_AND_ASSIGN(auto base_snapshot_id,
+                             writer->GetBaseSnapshotId({{"dt", "empty"}}, 0));
+        ASSERT_EQ(base_snapshot_id, std::nullopt);
+
+        ASSERT_OK(seed_snapshot(0, 1));
+        ASSERT_EQ(catalog->AcceptedSnapshots().back().Id(), 1);
+        ASSERT_OK_AND_ASSIGN(batch, make_batch("a", 0, R"([["a", 2]])"));
+        ASSERT_OK(writer->Write(std::move(batch)));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({{"dt", "a"}}, 0));
+        ASSERT_EQ(base_snapshot_id, 1);
+
+        ASSERT_OK(seed_snapshot(1, 2));
+        ASSERT_EQ(catalog->AcceptedSnapshots().back().Id(), 2);
+        ASSERT_OK_AND_ASSIGN(batch, make_batch("a", 1, R"([["a", 3]])"));
+        ASSERT_OK(writer->Write(std::move(batch)));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({{"dt", "a"}}, 1));
+        ASSERT_EQ(base_snapshot_id, 2);
+        ASSERT_OK_AND_ASSIGN(batch, make_batch("b", 0, R"([["b", 4]])"));
+        ASSERT_OK(writer->Write(std::move(batch)));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({{"dt", "b"}}, 0));
+        ASSERT_EQ(base_snapshot_id, 2);
+
+        ASSERT_OK_AND_ASSIGN(auto ignoring_writer, create_writer(true));
+        ASSERT_OK_AND_ASSIGN(batch, make_batch("a", 0, R"([["a", 5]])"));
+        ASSERT_OK(ignoring_writer->Write(std::move(batch)));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id,
+                             ignoring_writer->GetBaseSnapshotId({{"dt", "a"}}, 0));
+        ASSERT_EQ(base_snapshot_id, std::nullopt);
+        ASSERT_OK(ignoring_writer->Close());
+
+        const size_t snapshot_loads = catalog->LoadSnapshotCalls();
+        catalog->SetLoadSnapshotStatus(Status::IOError("catalog unavailable"));
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({{"dt", "a"}}, 0));
+        ASSERT_EQ(base_snapshot_id, 1);
+        ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({{"dt", "empty"}}, 0));
+        ASSERT_EQ(base_snapshot_id, std::nullopt);
+        ASSERT_NOK_WITH_MSG(writer->GetBaseSnapshotId({{"dt", "b"}}, 1), "writer does not exist");
+        ASSERT_EQ(catalog->LoadSnapshotCalls(), snapshot_loads);
+        ASSERT_OK(writer->Close());
+        ASSERT_NOK_WITH_MSG(writer->GetBaseSnapshotId({{"dt", "a"}}, 0), "writer does not exist");
+    }
+}
+
 TEST(FileStoreWriteTest, TestCreateWithInvalidInput) {
     auto dir = UniqueTestDirectory::Create();
     WriteContextBuilder builder(dir->Str(), "commit_user_1");
diff --git a/src/paimon/core/realtime/realtime_context_impl.cpp b/src/paimon/core/realtime/realtime_context_impl.cpp
index 00de0c35f..3a63948eb 100644
--- a/src/paimon/core/realtime/realtime_context_impl.cpp
+++ b/src/paimon/core/realtime/realtime_context_impl.cpp
@@ -73,6 +73,7 @@ struct RealtimeContextImpl::RefreshNotification {
     int64_t committed_end_offset;
     std::vector<std::shared_ptr<DataFileMeta>> active_data_files;
     std::optional<RealtimeDeduplicateState::PreparedSnapshot> prepared;
+    bool installed = false;
 };
 
 struct RealtimeContextImpl::RefreshState {
@@ -168,6 +169,17 @@ std::shared_ptr<Metrics> RealtimeContextImpl::GetMetrics() const {
     return metrics;
 }
 
+Result<std::optional<int64_t>> RealtimeContextImpl::GetRefreshedSnapshotId(
+    const RealtimePartitionBucket& partition_bucket) const {
+    std::lock_guard<std::mutex> progress_lock(progress_mutex_);
+    std::lock_guard<std::mutex> registry_lock(mutex_);
+    auto iter = stores_.find(partition_bucket);
+    if (iter == stores_.end()) {
+        return Status::KeyError("real-time partition-bucket store does not exist");
+    }
+    return iter->second.refreshed_snapshot_id;
+}
+
 Result<RealtimeStoreState> RealtimeContextImpl::GetOrCreateRealtimeStore(
     RealtimeStoreCreateRequest&& request, const RealtimePartitionBucket& partition_bucket) {
     PAIMON_RETURN_NOT_OK(CheckUsable());
@@ -259,9 +271,11 @@ Result<RealtimeStoreState> RealtimeContextImpl::GetOrCreateRealtimeStore(
                                    GetArrowPool(memory_pool)));
         store = std::move(deduplicate_store);
     }
-    stores_.emplace(partition_bucket,
-                    StoreEntry{store, requested_schema, mode, std::move(temp_directory),
-                               std::move(deduplicate_key_fields), deduplicate_state});
+    stores_.emplace(
+        partition_bucket,
+        StoreEntry{store, requested_schema, mode, std::move(temp_directory),
+                   std::move(deduplicate_key_fields), deduplicate_state,
+                   /*materialized_max_sequence_number=*/-1, last_refreshed_snapshot_id_});
     if (offset_iter != committed_offsets_.end()) {
         reclaimed_offsets_.emplace(partition_bucket, offset_iter->second);
     }
@@ -504,7 +518,7 @@ Status RealtimeContextImpl::InstallCommittedRefresh(int64_t snapshot_id, Refresh
     committed_data_files_ = std::move(refresh->next_data_files);
     last_refreshed_snapshot_id_ = snapshot_id;
     Status first_error = Status::OK();
-    for (const auto& [partition_bucket, notification] : refresh->notifications) {
+    for (auto& [partition_bucket, notification] : refresh->notifications) {
         Status status;
         if (notification.deduplicate_state) {
             Result<std::shared_ptr<RealtimeReadView>> retired =
@@ -527,6 +541,7 @@ Status RealtimeContextImpl::InstallCommittedRefresh(int64_t snapshot_id, Refresh
             status = notification.store->AdvanceCommittedOffset(notification.committed_end_offset);
         }
         if (status.ok()) {
+            notification.installed = true;
             reclaimed_offsets_[partition_bucket] = notification.committed_end_offset;
             if (notification.deduplicate_state) {
                 installed_snapshot_ids_[partition_bucket] = snapshot_id;
@@ -535,6 +550,15 @@ Status RealtimeContextImpl::InstallCommittedRefresh(int64_t snapshot_id, Refresh
             first_error = std::move(status);
         }
     }
+    {
+        std::lock_guard<std::mutex> registry_lock(mutex_);
+        for (auto& [partition_bucket, entry] : stores_) {
+            auto notification = refresh->notifications.find(partition_bucket);
+            if (notification == refresh->notifications.end() || notification->second.installed) {
+                entry.refreshed_snapshot_id = snapshot_id;
+            }
+        }
+    }
     return first_error;
 }
 
diff --git a/src/paimon/core/realtime/realtime_context_impl.h b/src/paimon/core/realtime/realtime_context_impl.h
index b7a0785c3..586ee87ab 100644
--- a/src/paimon/core/realtime/realtime_context_impl.h
+++ b/src/paimon/core/realtime/realtime_context_impl.h
@@ -91,6 +91,10 @@ class PAIMON_EXPORT RealtimeContextImpl final : public RealtimeContext {
 
     std::shared_ptr<Metrics> GetMetrics() const override;
 
+    // Snapshot last applied to this store, including initialization from committed progress.
+    Result<std::optional<int64_t>> GetRefreshedSnapshotId(
+        const RealtimePartitionBucket& partition_bucket) const;
+
     Result<RealtimeStoreState> GetOrCreateRealtimeStore(
         RealtimeStoreCreateRequest&& request, const RealtimePartitionBucket& partition_bucket);
 
@@ -127,6 +131,7 @@ class PAIMON_EXPORT RealtimeContextImpl final : public RealtimeContext {
         std::vector<std::string> deduplicate_key_fields;
         std::shared_ptr<RealtimeDeduplicateState> deduplicate_state;
         int64_t materialized_max_sequence_number = -1;
+        std::optional<int64_t> refreshed_snapshot_id;
     };
 
     struct RefreshNotification;
@@ -150,7 +155,7 @@ class PAIMON_EXPORT RealtimeContextImpl final : public RealtimeContext {
     std::shared_ptr<RealtimeStoreFactory> factory_;
     mutable std::mutex mutex_;
     std::mutex refresh_mutex_;
-    std::mutex progress_mutex_;
+    mutable std::mutex progress_mutex_;
     std::map<RealtimePartitionBucket, StoreEntry> stores_;
     // Full-table progress used as the initial offset when a store is created lazily.
     RealtimeOffsetMap committed_offsets_;
diff --git a/src/paimon/core/realtime/realtime_context_test.cpp b/src/paimon/core/realtime/realtime_context_test.cpp
index 87d0cb104..5a67ce8bc 100644
--- a/src/paimon/core/realtime/realtime_context_test.cpp
+++ b/src/paimon/core/realtime/realtime_context_test.cpp
@@ -294,9 +294,13 @@ TEST_F(RealtimeRefreshTest, TestSerializesConcurrentRefreshes) {
         return context_->AdvanceCommittedProgress(3, {{bucket_, 2}}, {{bucket_, {second}}});
     });
     ASSERT_EQ(std::future_status::timeout, next.wait_for(std::chrono::milliseconds(100)));
+    ASSERT_OK_AND_ASSIGN(auto snapshot_id, context_->GetRefreshedSnapshotId(bucket_));
+    ASSERT_EQ(snapshot_id, 1);
     open_gate->Release();
     ASSERT_OK(refreshing.get());
     ASSERT_OK(next.get());
+    ASSERT_OK_AND_ASSIGN(snapshot_id, context_->GetRefreshedSnapshotId(bucket_));
+    ASSERT_EQ(snapshot_id, 3);
     ASSERT_OK_AND_ASSIGN(RealtimeReadState after, context_->AcquireReadState());
     ASSERT_EQ(2, after.committed_offsets.at(bucket_));
     ASSERT_EQ(std::vector<std::shared_ptr<DataFileMeta>>{second},
@@ -305,6 +309,33 @@ TEST_F(RealtimeRefreshTest, TestSerializesConcurrentRefreshes) {
                         "committed snapshot cannot move backwards");
 }
 
+TEST_F(RealtimeRefreshTest, TestBaseSnapshotTracksCompactionRefreshAndRetry) {
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> original,
+                         CreateFile("original.parquet", R"([[0, 1, "value"]])"));
+    ASSERT_OK(context_->AdvanceCommittedProgress(2, {{bucket_, 1}}, {{bucket_, {original}}}));
+    ASSERT_OK_AND_ASSIGN(auto snapshot_id, context_->GetRefreshedSnapshotId(bucket_));
+    ASSERT_EQ(snapshot_id, 2);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> compacted,
+                         CreateFile("compacted.parquet", R"([[0, 1, "value"]])"));
+    auto open_gate = fs_->BlockNext(BlockingFileSystem::Operation::OPEN,
+                                    path_factory_->ToPath(compacted) + ".offset");
+    open_gate->Release(Status::IOError("injected compaction sidecar failure"));
+    ASSERT_NOK_WITH_MSG(
+        context_->AdvanceCommittedProgress(3, {{bucket_, 1}}, {{bucket_, {compacted}}}),
+        "injected compaction sidecar failure");
+    ASSERT_OK_AND_ASSIGN(snapshot_id, context_->GetRefreshedSnapshotId(bucket_));
+    ASSERT_EQ(snapshot_id, 2);
+    ASSERT_EQ(std::vector<std::shared_ptr<DataFileMeta>>{original},
+              state_.deduplicate_state->committed_file_lookup_->DataFiles());
+
+    ASSERT_OK(context_->AdvanceCommittedProgress(3, {{bucket_, 1}}, {{bucket_, {compacted}}}));
+    ASSERT_OK_AND_ASSIGN(snapshot_id, context_->GetRefreshedSnapshotId(bucket_));
+    ASSERT_EQ(snapshot_id, 3);
+    ASSERT_EQ(std::vector<std::shared_ptr<DataFileMeta>>{compacted},
+              state_.deduplicate_state->committed_file_lookup_->DataFiles());
+}
+
 TEST(RealtimeContextTest, TestReusesStoreAndCapturesRegisteredViews) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
@@ -634,6 +665,7 @@ TEST(RealtimeContextTest, TestRetriesOnlyIncompleteReclamation) {
     auto factory = std::make_shared<TestingRealtimeStoreFactory>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context, CreateContext(factory));
     const std::map<std::string, std::string> partition = {{"dt", "2026-08-02"}};
+    ASSERT_OK(context->AdvanceCommittedProgress(4, {}, {}));
 
     ASSERT_OK(
         GetOrCreateAppendStore(context, partition, 0, MakeWriteSchema(), {}, GetDefaultPool()));
@@ -654,6 +686,11 @@ TEST(RealtimeContextTest, TestRetriesOnlyIncompleteReclamation) {
     ASSERT_EQ(std::vector<int64_t>({7}), factory->stores[0]->committed_offsets);
     ASSERT_TRUE(factory->stores[1]->committed_offsets.empty());
     ASSERT_EQ(std::vector<int64_t>({9}), factory->stores[2]->committed_offsets);
+    for (int32_t bucket = 0; bucket < 3; ++bucket) {
+        ASSERT_OK_AND_ASSIGN(auto snapshot_id, context->GetRefreshedSnapshotId(
+                                                   RealtimePartitionBucket(partition, bucket)));
+        ASSERT_EQ(snapshot_id, bucket == 1 ? 4 : 5);
+    }
 
     ASSERT_OK_AND_ASSIGN(
         RealtimeStoreState failed_store_state,
@@ -665,6 +702,21 @@ TEST(RealtimeContextTest, TestRetriesOnlyIncompleteReclamation) {
     ASSERT_EQ(2, factory->stores[1]->advance_count);
     ASSERT_EQ(1, factory->stores[2]->advance_count);
     ASSERT_EQ(std::vector<int64_t>({8}), factory->stores[1]->committed_offsets);
+    for (int32_t bucket = 0; bucket < 3; ++bucket) {
+        ASSERT_OK_AND_ASSIGN(auto snapshot_id, context->GetRefreshedSnapshotId(
+                                                   RealtimePartitionBucket(partition, bucket)));
+        ASSERT_EQ(snapshot_id, 5);
+    }
+    // A snapshot with unchanged offsets still becomes the current base without re-retiring rows.
+    ASSERT_OK(context->AdvanceCommittedProgress(6, committed_offsets, {}));
+    ASSERT_EQ(1, factory->stores[0]->advance_count);
+    ASSERT_EQ(2, factory->stores[1]->advance_count);
+    ASSERT_EQ(1, factory->stores[2]->advance_count);
+    for (int32_t bucket = 0; bucket < 3; ++bucket) {
+        ASSERT_OK_AND_ASSIGN(auto snapshot_id, context->GetRefreshedSnapshotId(
+                                                   RealtimePartitionBucket(partition, bucket)));
+        ASSERT_EQ(snapshot_id, 6);
+    }
 }
 
 TEST(RealtimeContextTest, TestRequiresReopenWhenCommittedProgressMovesBackwards) {
diff --git a/src/paimon/core/table/format/format_table_file_store_write.cpp b/src/paimon/core/table/format/format_table_file_store_write.cpp
index 50d417db4..1d888402a 100644
--- a/src/paimon/core/table/format/format_table_file_store_write.cpp
+++ b/src/paimon/core/table/format/format_table_file_store_write.cpp
@@ -92,4 +92,9 @@ Status FormatTableFileStoreWrite::Close() {
     return Status::OK();
 }
 
+Result<std::optional<int64_t>> FormatTableFileStoreWrite::GetBaseSnapshotId(
+    const std::map<std::string, std::string>&, int32_t) const {
+    return Status::Invalid("format tables do not have base snapshots");
+}
+
 }  // namespace paimon
diff --git a/src/paimon/core/table/format/format_table_file_store_write.h b/src/paimon/core/table/format/format_table_file_store_write.h
index 2e6f49c06..9ef37c974 100644
--- a/src/paimon/core/table/format/format_table_file_store_write.h
+++ b/src/paimon/core/table/format/format_table_file_store_write.h
@@ -21,6 +21,7 @@
 #include <cstdint>
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -59,6 +60,9 @@ class FormatTableFileStoreWrite : public FileStoreWrite {
 
     std::shared_ptr<Metrics> GetMetrics() const override;
 
+    Result<std::optional<int64_t>> GetBaseSnapshotId(
+        const std::map<std::string, std::string>& partition, int32_t bucket) const override;
+
     Status Close() override;
 
  private:
diff --git a/src/paimon/core/table/format/format_table_test.cpp b/src/paimon/core/table/format/format_table_test.cpp
index e8391f68c..40e4c3ebd 100644
--- a/src/paimon/core/table/format/format_table_test.cpp
+++ b/src/paimon/core/table/format/format_table_test.cpp
@@ -2678,6 +2678,8 @@ TEST(FormatTableTest, TestTheGenericEntryPointsReachAFormatTable) {
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context, write_builder.Finish());
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> write,
                          FileStoreWrite::Create(std::move(write_context)));
+    ASSERT_NOK_WITH_MSG(write->GetBaseSnapshotId({{"dt", "20240101"}}, 0),
+                        "format tables do not have base snapshots");
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
                          MakeBatch({1, 2}, {"alice", "bob"}, "20240101", {{"dt", "20240101"}}));
     ASSERT_OK(write->Write(std::move(batch)));
diff --git a/test/inte/realtime_write_inte_test.cpp b/test/inte/realtime_write_inte_test.cpp
index e520d43e9..2e6339cd4 100644
--- a/test/inte/realtime_write_inte_test.cpp
+++ b/test/inte/realtime_write_inte_test.cpp
@@ -1616,6 +1616,8 @@ TEST_F(RealtimeWriteInteTest, TestRefreshPreparationFailureKeepsProgressAndCanRe
     open_gate->Release(Status::IOError("injected sidecar open failure"));
     ASSERT_NOK_WITH_MSG(writer->RefreshCommittedSnapshot(snapshot),
                         "injected sidecar open failure");
+    ASSERT_OK_AND_ASSIGN(auto base_snapshot_id, writer->GetBaseSnapshotId({}, 0));
+    ASSERT_EQ(base_snapshot_id, std::nullopt);
     ASSERT_OK_AND_ASSIGN(RealtimeMetricValues failed, ReadRealtimeMetrics(writer->GetMetrics()));
     ASSERT_EQ(before.sealed_rows, failed.sealed_rows);
     ASSERT_EQ(before.total_memory, failed.total_memory);
@@ -1623,6 +1625,8 @@ TEST_F(RealtimeWriteInteTest, TestRefreshPreparationFailureKeepsProgressAndCanRe
     ASSERT_EQ(initial, visible);
 
     ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot));
+    ASSERT_OK_AND_ASSIGN(base_snapshot_id, writer->GetBaseSnapshotId({}, 0));
+    ASSERT_EQ(base_snapshot_id, snapshot);
     ASSERT_OK_AND_ASSIGN(RealtimeMetricValues after, ReadRealtimeMetrics(writer->GetMetrics()));
     ASSERT_EQ(0, after.sealed_rows);
     ASSERT_EQ(0, after.total_rows);