diff --git a/velox/connectors/Connector.h b/velox/connectors/Connector.h
index 37ee5d02505..96720b6249c 100644
--- a/velox/connectors/Connector.h
+++ b/velox/connectors/Connector.h
@@ -242,6 +242,10 @@ class DataSink {
 
   /// Updates the current event-time watermark in milliseconds.
   virtual void setWatermark(int64_t watermark) {}
+
+  /// Restores sink state from checkpoint records produced by snapshot().
+  virtual void restoreState(
+      const std::vector<std::string>& /*checkpointRecords*/) {}
 };
 
 class DataSource {
diff --git a/velox/connectors/filesystem/FileSystemDataSink.cpp b/velox/connectors/filesystem/FileSystemDataSink.cpp
index bfa19fc6644..77eefe46c36 100644
--- a/velox/connectors/filesystem/FileSystemDataSink.cpp
+++ b/velox/connectors/filesystem/FileSystemDataSink.cpp
@@ -16,6 +16,7 @@
 
 #include "velox/connectors/filesystem/FileSystemDataSink.h"
 #include <common/compression/Compression.h>
+#include <folly/json.h>
 #include <algorithm>
 #include "boost/uuid/uuid.hpp"
 #include "boost/uuid/uuid_generators.hpp"
@@ -90,15 +91,12 @@ FileSystemDataSink::FileSystemDataSink(
       dataChannels_(
           getNonPartitionChannels(partitionChannels_, inputType_->size())),
       writerFactory_(dwio::common::getWriterFactory(writeConfig_->getFormat())),
-      fileNameGenerator_(
-          std::make_shared<const FsFileNameGenerator>(
-              queryCtx_->sessionProperties()->get<std::string>(
-                  "query_uuid",
-                  ""),
-              "",
-              queryCtx_->sessionProperties()->get<std::string>(
-                  "task_index",
-                  "0"))) {}
+      fileNameGenerator_(std::make_shared<FsFileNameGenerator>(
+          queryCtx_->sessionProperties()->get<std::string>("query_uuid", ""),
+          "",
+          queryCtx_->sessionProperties()->get<std::string>(
+              "task_index",
+              "0"))) {}
 
 const std::unique_ptr<dwio::common::Writer> FileSystemDataSink::createWriter(
     const std::string& writePath,
@@ -148,8 +146,8 @@ const std::unique_ptr<dwio::common::Writer> FileSystemDataSink::createWriter(
 }
 
 const std::pair<std::string, std::string>
-FileSystemDataSink::getWriterFileNames() const {
-  return fileNameGenerator_->gen();
+FileSystemDataSink::getWriterFileNames(const std::string& bucketId) {
+  return fileNameGenerator_->gen(bucketId);
 }
 
 namespace {
@@ -179,11 +177,55 @@ std::string makePartitionDirectory(
   }
   return (fs::path(tableDirectory) / partitionSubdirectory.value()).string();
 }
+
+void validatePendingFileState(const folly::dynamic& pendingFile) {
+  VELOX_USER_CHECK(
+      pendingFile.isObject(),
+      "Invalid filesystem checkpoint state: pending file must be an object");
+  const auto requireInt = [&](const char* key) {
+    VELOX_USER_CHECK(
+        pendingFile.count(key) && pendingFile[key].isInt(),
+        "Invalid filesystem checkpoint state: pending file missing int field "
+        "'{}'",
+        key);
+  };
+  const auto requireBool = [&](const char* key) {
+    VELOX_USER_CHECK(
+        pendingFile.count(key) && pendingFile[key].isBool(),
+        "Invalid filesystem checkpoint state: pending file missing bool field "
+        "'{}'",
+        key);
+  };
+  const auto requireString = [&](const char* key) {
+    VELOX_USER_CHECK(
+        pendingFile.count(key) && pendingFile[key].isString(),
+        "Invalid filesystem checkpoint state: pending file missing string "
+        "field '{}'",
+        key);
+  };
+  requireInt("checkpointId");
+  requireBool("hasPartition");
+  requireString("partitionName");
+  requireString("targetFileName");
+  requireString("targetDirectory");
+  requireString("writeFileName");
+  requireString("writeDirectory");
+  requireInt("createTime");
+  requireBool("committed");
+  requireBool("fileRolled");
+}
+
 } // namespace
 
-FsWriterParameters FileSystemDataSink::getWriterParameters(
+std::string FileSystemDataSink::bucketIdForPartition(
     const std::optional<std::string>& partition) const {
-  auto [targetFileName, writeFileName] = getWriterFileNames();
+  return partition.value_or("");
+}
+
+FsWriterParameters FileSystemDataSink::getWriterParameters(
+    const std::optional<std::string>& partition) {
+  auto [targetFileName, writeFileName] =
+      getWriterFileNames(bucketIdForPartition(partition));
   return FsWriterParameters{
       partition,
       targetFileName,
@@ -554,7 +596,125 @@ std::vector<std::string> FileSystemDataSink::snapshot(int64_t checkpointId) {
         pendingWriterInfo_.end());
     pendingWriterInfo_.clear();
   }
-  return {};
+  return snapshotBucketState(checkpointId);
+}
+
+std::vector<std::string> FileSystemDataSink::snapshotBucketState(
+    int64_t checkpointId) const {
+  folly::dynamic state = folly::dynamic::object;
+  state["connector"] = "filesystem";
+  state["checkpointId"] = checkpointId;
+  // StreamingFileSink stores maxPartCounter in a separate partCounterState.
+  state["partCounter"] = fileNameGenerator_->maxPartCounter();
+  state["pendingFiles"] = folly::dynamic::array;
+
+  for (const auto& [pendingCheckpointId, writerInfos] : checkpointWriterInfo_) {
+    for (const auto& writerInfo : writerInfos) {
+      const auto& writerParams = writerInfo->writerParameters;
+      folly::dynamic pendingFile = folly::dynamic::object;
+      pendingFile["checkpointId"] = pendingCheckpointId;
+      pendingFile["bucketId"] =
+          bucketIdForPartition(writerParams.partitionName());
+      pendingFile["partitionName"] = writerParams.partitionName().value_or("");
+      pendingFile["hasPartition"] = writerParams.partitionName().has_value();
+      pendingFile["targetFileName"] = writerParams.targetFileName();
+      pendingFile["targetDirectory"] = writerParams.targetDirectory();
+      pendingFile["writeFileName"] = writerParams.writeFileName();
+      pendingFile["writeDirectory"] = writerParams.writeDirectory();
+      pendingFile["createTime"] =
+          static_cast<int64_t>(writerInfo->createTime());
+      pendingFile["committed"] = writerInfo->isCommitted();
+      pendingFile["fileRolled"] = writerInfo->isFileRolled();
+      state["pendingFiles"].push_back(std::move(pendingFile));
+    }
+  }
+
+  return {folly::toJson(state)};
+}
+
+void FileSystemDataSink::restoreState(
+    const std::vector<std::string>& checkpointRecords) {
+  for (const auto& checkpointRecord : checkpointRecords) {
+    folly::dynamic state;
+    try {
+      state = folly::parseJson(checkpointRecord);
+    } catch (const std::exception&) {
+      continue;
+    }
+    if (!state.isObject() || !state.count("connector") ||
+        !state["connector"].isString() ||
+        state["connector"].asString() != "filesystem") {
+      continue;
+    }
+
+    // Prefer StreamingFileSink-style global max partCounter. Fall back to the
+    // legacy per-bucket buckets[].partCounter format by taking the max.
+    if (state.count("partCounter")) {
+      VELOX_USER_CHECK(
+          state["partCounter"].isInt(),
+          "Invalid filesystem checkpoint state: partCounter must be an int");
+      const auto partCounter = state["partCounter"].asInt();
+      VELOX_USER_CHECK_GE(partCounter, 0, "Invalid restored part counter");
+      fileNameGenerator_->restoreMaxPartCounter(
+          static_cast<uint64_t>(partCounter));
+    } else if (state.count("buckets")) {
+      VELOX_USER_CHECK(
+          state["buckets"].isArray(),
+          "Invalid filesystem checkpoint state: buckets must be an array");
+      for (const auto& bucket : state["buckets"]) {
+        VELOX_USER_CHECK(
+            bucket.isObject() && bucket.count("partCounter") &&
+                bucket["partCounter"].isInt(),
+            "Invalid filesystem checkpoint state: each bucket must be an "
+            "object with an int partCounter");
+        const auto partCounter = bucket["partCounter"].asInt();
+        VELOX_USER_CHECK_GE(partCounter, 0, "Invalid restored part counter");
+        fileNameGenerator_->restoreMaxPartCounter(
+            static_cast<uint64_t>(partCounter));
+      }
+    }
+
+    if (!state.count("pendingFiles")) {
+      continue;
+    }
+    VELOX_USER_CHECK(
+        state["pendingFiles"].isArray(),
+        "Invalid filesystem checkpoint state: pendingFiles must be an array");
+    int32_t restoredFileIndex = 0;
+    for (const auto& pendingFile : state["pendingFiles"]) {
+      validatePendingFileState(pendingFile);
+      const auto pendingCheckpointId = pendingFile["checkpointId"].asInt();
+      const bool hasPartition = pendingFile["hasPartition"].asBool();
+      std::optional<std::string> partitionName = std::nullopt;
+      if (hasPartition) {
+        partitionName = pendingFile["partitionName"].asString();
+        partitionCreateTimeMs_.try_emplace(
+            partitionName.value(), pendingFile["createTime"].asInt() * 1000);
+      }
+
+      auto* connectorPool = queryCtx_->connectorMemoryPool();
+      auto writerPool = connectorPool->addAggregateChild(fmt::format(
+          "{}.restored-file-system-sink.{}.{}",
+          connectorPool->name(),
+          pendingCheckpointId,
+          restoredFileIndex++));
+      auto sinkPool = createSinkPool(writerPool);
+      auto writerInfo = std::make_shared<FsWriterInfo>(
+          FsWriterParameters{
+              partitionName,
+              pendingFile["targetFileName"].asString(),
+              pendingFile["targetDirectory"].asString(),
+              pendingFile["writeFileName"].asString(),
+              pendingFile["writeDirectory"].asString()},
+          std::move(writerPool),
+          std::move(sinkPool),
+          static_cast<time_t>(pendingFile["createTime"].asInt()));
+      writerInfo->setCommitted(pendingFile["committed"].asBool());
+      writerInfo->setFileRolled(pendingFile["fileRolled"].asBool());
+      checkpointWriterInfo_[pendingCheckpointId].emplace_back(
+          std::move(writerInfo));
+    }
+  }
 }
 
 // Renames in-progress files and returns paths ready for Flink-side partition
@@ -617,11 +777,21 @@ std::vector<std::string> FileSystemDataSink::commit(int64_t checkpointId) {
           fs_->rename(writeFileName, targetFileName);
           writerInfo->setFileRolled(true);
         } catch (const std::exception& e) {
-          VELOX_FAIL(
-              "Failed to rename file {} to target {}, exception: {}",
-              writeFileName,
-              targetFileName,
-              e.what());
+          const auto targetExists = fs_->exists(targetFileName);
+          const auto writeExists = fs_->exists(writeFileName);
+          if (targetExists && !writeExists) {
+            LOG(INFO) << "Treat rename as already completed for file "
+                      << writeFileName << " to target " << targetFileName
+                      << ", targetExists=" << targetExists
+                      << ", exception: " << e.what();
+            writerInfo->setFileRolled(true);
+          } else {
+            VELOX_FAIL(
+                "Failed to rename file {} to target {}, exception: {}",
+                writeFileName,
+                targetFileName,
+                e.what());
+          }
         }
       }
 
@@ -762,11 +932,18 @@ const int64_t FileSystemDataSink::extractTimestampFromPartitionName(
   return timestamp.getSeconds();
 }
 
-const std::pair<std::string, std::string> FsFileNameGenerator::gen() const {
+const std::pair<std::string, std::string> FsFileNameGenerator::gen(
+    const std::string& bucketId) {
+  // Initialize this bucket's counter from the global max, matching
+  // StreamingFileSink Buckets#getOrCreateBucketForBucketId.
+  auto [it, inserted] = partCounters_.try_emplace(bucketId, maxPartCounter_);
+  (void)inserted;
+  auto& partCounter = it->second;
+
   std::pair<std::string, std::string> fileNames;
   std::stringstream targetFileName;
   std::stringstream writeFileName;
-  targetFileName << "part-" << prefix_ << "-" << taskId_ << "-" << partCounter_
+  targetFileName << "part-" << prefix_ << "-" << taskId_ << "-" << partCounter
                  << suffix_;
   boost::uuids::random_generator generator;
   boost::uuids::uuid uuid = generator();
@@ -774,8 +951,20 @@ const std::pair<std::string, std::string> FsFileNameGenerator::gen() const {
                 << to_string(uuid);
   fileNames.first = targetFileName.str();
   fileNames.second = writeFileName.str();
-  partCounter_++;
+  partCounter++;
+  // Keep the global max in sync so inactive buckets reopened later do not
+  // restart from 0 and overwrite committed parts.
+  maxPartCounter_ = std::max(maxPartCounter_, partCounter);
   return fileNames;
 }
 
+void FsFileNameGenerator::restoreMaxPartCounter(uint64_t partCounter) {
+  maxPartCounter_ = std::max(maxPartCounter_, partCounter);
+  // Buckets created after restore start from the restored max, as in
+  // StreamingFileSink Buckets#initializePartCounter / restoreBucket.
+  for (auto& [_, counter] : partCounters_) {
+    counter = std::max(counter, maxPartCounter_);
+  }
+}
+
 } // namespace facebook::velox::connector::filesystem
diff --git a/velox/connectors/filesystem/FileSystemDataSink.h b/velox/connectors/filesystem/FileSystemDataSink.h
index 157a0897dd3..9bec4811c02 100644
--- a/velox/connectors/filesystem/FileSystemDataSink.h
+++ b/velox/connectors/filesystem/FileSystemDataSink.h
@@ -18,6 +18,8 @@
 #include <cstdint>
 #include <map>
 #include <optional>
+#include <string>
+#include <unordered_map>
 #include "boost/algorithm/string.hpp"
 #include "folly/container/F14Map.h"
 #include "velox/common/file/FileSystems.h"
@@ -82,11 +84,11 @@ struct FsWriterInfo : public hive::HiveWriterInfo {
     committed_ = committed;
   }
 
-  bool isCommitted() {
+  bool isCommitted() const {
     return committed_;
   }
 
-  bool isFileRolled() {
+  bool isFileRolled() const {
     return fileRolled_;
   }
 
@@ -94,6 +96,10 @@ struct FsWriterInfo : public hive::HiveWriterInfo {
     fileRolled_ = fileRolled;
   }
 
+  time_t createTime() const {
+    return createTime_;
+  }
+
  private:
   const time_t createTime_;
   bool committed_ = false;
@@ -102,6 +108,9 @@ struct FsWriterInfo : public hive::HiveWriterInfo {
 
 using FsWriterInfoPtr = std::shared_ptr<FsWriterInfo>;
 
+/// Generates part file names using a StreamingFileSink-style counter model:
+/// each active bucket keeps an in-memory partCounter initialized from the
+/// global maxPartCounter; only maxPartCounter is checkpointed.
 class FsFileNameGenerator {
  public:
   FsFileNameGenerator(
@@ -110,13 +119,27 @@ class FsFileNameGenerator {
       const std::string& taskId)
       : prefix_(prefix), suffix_(suffix), taskId_(taskId) {}
 
-  const std::pair<std::string, std::string> gen() const;
+  // Stateful: advances per-bucket and global part counters.
+  const std::pair<std::string, std::string> gen(const std::string& bucketId);
+
+  /// Global max part counter across all buckets (checkpointed like Flink's
+  /// partCounterState).
+  uint64_t maxPartCounter() const {
+    return maxPartCounter_;
+  }
+
+  /// Restores the global max part counter. New/restored buckets start from
+  /// this value so previously committed part files are not overwritten.
+  void restoreMaxPartCounter(uint64_t partCounter);
 
  private:
   const std::string prefix_;
   const std::string suffix_;
   const std::string taskId_;
-  mutable uint64_t partCounter_ = 0;
+  /// Max part counter used across all buckets for this sink task.
+  uint64_t maxPartCounter_{0};
+  /// Per-bucket counters for currently active buckets (not checkpointed).
+  std::unordered_map<std::string, uint64_t> partCounters_;
 };
 
 class FsPartitionIdGenerator : public hive::PartitionIdGenerator {
@@ -224,6 +247,9 @@ class FileSystemDataSink : public DataSink {
   /// Receives event-time watermark in milliseconds.
   void setWatermark(int64_t watermark) override;
 
+  /// Restores max part counter and pending files from checkpoint records.
+  void restoreState(const std::vector<std::string>& checkpointRecords) override;
+
   // For test.
   const std::vector<FsWriterInfoPtr>& getWriteInfos() {
     return writerInfo_;
@@ -280,7 +306,8 @@ class FileSystemDataSink : public DataSink {
   std::vector<vector_size_t*> rawPartitionRows_;
   std::vector<vector_size_t> partitionSizes_;
   const std::shared_ptr<dwio::common::WriterFactory> writerFactory_;
-  const std::shared_ptr<const FsFileNameGenerator> fileNameGenerator_;
+  // Non-const: file name generation mutates part counters.
+  const std::shared_ptr<FsFileNameGenerator> fileNameGenerator_;
   std::shared_ptr<filesystems::FileSystem> fs_;
   int64_t watermark_{INT64_MIN};
   // Partition creation time in milliseconds (system clock) for process-time
@@ -326,11 +353,18 @@ class FileSystemDataSink : public DataSink {
   // prefix to the target file for write file name. The coordinator (or driver
   // for Presto on spark) will rename the write file to target file to commit
   // the table write when update the metadata store.
-  const std::pair<std::string, std::string> getWriterFileNames() const;
+  // Non-const: delegates to stateful FsFileNameGenerator::gen().
+  const std::pair<std::string, std::string> getWriterFileNames(
+      const std::string& bucketId);
 
   FsWriterParameters getWriterParameters(
+      const std::optional<std::string>& partition);
+
+  std::string bucketIdForPartition(
       const std::optional<std::string>& partition) const;
 
+  std::vector<std::string> snapshotBucketState(int64_t checkpointId) const;
+
   // Get the HiveWriter corresponding to the row from partitionIds.
   FOLLY_ALWAYS_INLINE FsWriterId getWriterId(size_t row) const;
 
diff --git a/velox/connectors/filesystem/tests/FileSystemConnectorTest.cpp b/velox/connectors/filesystem/tests/FileSystemConnectorTest.cpp
index 7828f0333e5..d2d06969fd3 100644
--- a/velox/connectors/filesystem/tests/FileSystemConnectorTest.cpp
+++ b/velox/connectors/filesystem/tests/FileSystemConnectorTest.cpp
@@ -333,6 +333,95 @@ TEST_F(
   fs->remove(postBarrierTargetPath);
 }
 
+TEST_F(FileSystemConnectorTest, testRestoreBucketStateKeepsPartCounter) {
+  std::shared_ptr<connector::Connector> fsConnector =
+      getConnector(fileSystemConnectorId);
+  auto queryCtx = createQueryCtx();
+  std::unique_ptr<DataSink> dataSink = fsConnector->createDataSink(
+      inputType,
+      createFsTableHandle({}, {}),
+      queryCtx.get(),
+      CommitStrategy::kNoCommit);
+  auto* fsDataSink = reinterpret_cast<FileSystemDataSink*>(dataSink.get());
+  ASSERT_TRUE(fsDataSink != nullptr);
+
+  fsDataSink->appendData(createSingleInputRow());
+  const auto checkpointTargetPath =
+      fsDataSink->getWriteInfos()[0]->writerParameters.targetDirectory() + "/" +
+      fsDataSink->getWriteInfos()[0]->writerParameters.targetFileName();
+  const std::vector<std::string> checkpointRecords = fsDataSink->snapshot(10);
+  ASSERT_EQ(checkpointRecords.size(), 1);
+  // StreamingFileSink-style state: global max partCounter, not per-bucket map.
+  ASSERT_NE(checkpointRecords[0].find("\"partCounter\""), std::string::npos);
+  ASSERT_EQ(checkpointRecords[0].find("\"buckets\""), std::string::npos);
+  dataSink.reset();
+
+  auto restoredQueryCtx = createQueryCtx();
+  std::unique_ptr<DataSink> restoredDataSink = fsConnector->createDataSink(
+      inputType,
+      createFsTableHandle({}, {}),
+      restoredQueryCtx.get(),
+      CommitStrategy::kNoCommit);
+  auto* restoredFsDataSink =
+      reinterpret_cast<FileSystemDataSink*>(restoredDataSink.get());
+  ASSERT_TRUE(restoredFsDataSink != nullptr);
+  restoredFsDataSink->restoreState(checkpointRecords);
+
+  restoredFsDataSink->appendData(createSingleInputRow());
+  const auto restoredTargetPath = restoredFsDataSink->getWriteInfos()[0]
+                                      ->writerParameters.targetDirectory() +
+      "/" +
+      restoredFsDataSink->getWriteInfos()[0]->writerParameters.targetFileName();
+  ASSERT_NE(checkpointTargetPath, restoredTargetPath);
+
+  const std::vector<std::string> committed = restoredFsDataSink->commit(10);
+  ASSERT_EQ(committed.size(), 1);
+  ASSERT_EQ(committed[0], checkpointTargetPath);
+
+  auto fs =
+      filesystems::getFileSystem(dataPath, fsConnector->connectorConfig());
+  ASSERT_TRUE(fs->exists(checkpointTargetPath));
+  fs->remove(checkpointTargetPath);
+
+  restoredFsDataSink->snapshot(11);
+  const std::vector<std::string> committed2 = restoredFsDataSink->commit(11);
+  ASSERT_EQ(committed2.size(), 1);
+  ASSERT_EQ(committed2[0], restoredTargetPath);
+  ASSERT_TRUE(fs->exists(restoredTargetPath));
+  fs->remove(restoredTargetPath);
+}
+
+TEST_F(FileSystemConnectorTest, testRestoreLegacyPerBucketPartCounterState) {
+  std::shared_ptr<connector::Connector> fsConnector =
+      getConnector(fileSystemConnectorId);
+  auto restoredQueryCtx = createQueryCtx();
+  std::unique_ptr<DataSink> restoredDataSink = fsConnector->createDataSink(
+      inputType,
+      createFsTableHandle({}, {}),
+      restoredQueryCtx.get(),
+      CommitStrategy::kNoCommit);
+  auto* restoredFsDataSink =
+      reinterpret_cast<FileSystemDataSink*>(restoredDataSink.get());
+  ASSERT_TRUE(restoredFsDataSink != nullptr);
+
+  // Legacy state format with per-bucket counters; restore should take the max.
+  const std::string legacyState = R"({
+    "connector":"filesystem",
+    "checkpointId":1,
+    "buckets":[
+      {"bucketId":"","partCounter":3},
+      {"bucketId":"p=0","partCounter":5}
+    ],
+    "pendingFiles":[]
+  })";
+  restoredFsDataSink->restoreState({legacyState});
+  restoredFsDataSink->appendData(createSingleInputRow());
+  const auto targetFileName =
+      restoredFsDataSink->getWriteInfos()[0]->writerParameters.targetFileName();
+  // After restoring max=5, the next part file must use counter 5.
+  ASSERT_NE(targetFileName.find("-5"), std::string::npos);
+}
+
 TEST_F(FileSystemConnectorTest, testNonPartitionedFileCommit) {
   std::shared_ptr<connector::Connector> fsConnector =
       getConnector(fileSystemConnectorId);
diff --git a/velox/exec/TableWriter.h b/velox/exec/TableWriter.h
index 9fa8664d072..810364a53f7 100644
--- a/velox/exec/TableWriter.h
+++ b/velox/exec/TableWriter.h
@@ -132,6 +132,13 @@ class TableWriter : public Operator {
     }
   }
 
+  void restoreState(
+      const std::vector<std::string>& checkpointRecords) override {
+    if (dataSink_) {
+      dataSink_->restoreState(checkpointRecords);
+    }
+  }
+
   virtual bool needsInput() const override {
     return true;
   }