apache · platinumhamburg · Jul 6, 2026
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -57,6 +57,8 @@
 import org.apache.fluss.exception.TooManyPartitionsException;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.fs.FsPathAndFileName;
+import org.apache.fluss.kv.snapshot.CompletedSnapshot;
+import org.apache.fluss.kv.snapshot.KvSnapshotHandle;
 import org.apache.fluss.metadata.AggFunctions;
 import org.apache.fluss.metadata.DatabaseChange;
 import org.apache.fluss.metadata.DatabaseDescriptor;
@@ -77,8 +79,6 @@
 import org.apache.fluss.rpc.messages.ListOffsetsRequest;
 import org.apache.fluss.rpc.messages.ListOffsetsResponse;
 import org.apache.fluss.server.coordinator.CoordinatorContext;
-import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
-import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle;
 import org.apache.fluss.server.log.LogTablet;
 import org.apache.fluss.server.metadata.ServerInfo;
 import org.apache.fluss.server.replica.Replica;

diff --git a/...uss/server/kv/autoinc/AutoIncIDRange.java → ...ache/fluss/kv/autoinc/AutoIncIDRange.java b/...uss/server/kv/autoinc/AutoIncIDRange.java → ...ache/fluss/kv/autoinc/AutoIncIDRange.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.fluss.server.kv.autoinc;
+package org.apache.fluss.kv.autoinc;
 
 import java.util.Objects;
 

diff --git a/...server/kv/snapshot/CompletedSnapshot.java → .../fluss/kv/snapshot/CompletedSnapshot.java b/...server/kv/snapshot/CompletedSnapshot.java → .../fluss/kv/snapshot/CompletedSnapshot.java
@@ -15,13 +15,13 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
+package org.apache.fluss.kv.snapshot;
 
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.fs.FileSystem;
 import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.kv.autoinc.AutoIncIDRange;
 import org.apache.fluss.metadata.TableBucket;
-import org.apache.fluss.server.kv.autoinc.AutoIncIDRange;
 import org.apache.fluss.utils.concurrent.FutureUtils;
 
 import javax.annotation.Nullable;
@@ -106,7 +106,7 @@ public CompletedSnapshot(
     }
 
     @VisibleForTesting
-    CompletedSnapshot(
+    public CompletedSnapshot(
             TableBucket tableBucket,
             long snapshotID,
             FsPath snapshotLocation,
@@ -152,22 +152,11 @@ public long getSnapshotSize() {
         return kvSnapshotHandle.getSnapshotSize();
     }
 
-    /**
-     * Register all shared kv files in the given registry. This method is called before the snapshot
-     * is added into the store.
-     *
-     * @param sharedKvFileRegistry The registry where shared kv files are registered
-     */
-    public void registerSharedKvFilesAfterRestored(SharedKvFileRegistry sharedKvFileRegistry) {
-        sharedKvFileRegistry.registerAllAfterRestored(this);
-    }
-
     public CompletableFuture<Void> discardAsync(Executor ioExecutor) {
-        // it'll discard the snapshot files for kv, it'll always discard
-        // the private files; for shared files, only if they're not be registered in
-        // SharedKvRegistry, can the files be deleted.
+        // Discard private files only; shared files are managed by the SharedKvFileRegistry
+        // on the server side.
         CompletableFuture<Void> discardKvFuture =
-                FutureUtils.runAsync(kvSnapshotHandle::discard, ioExecutor);
+                FutureUtils.runAsync(kvSnapshotHandle::discardPrivateFiles, ioExecutor);
 
         CompletableFuture<Void> discardMetaFileFuture =
                 FutureUtils.runAsync(this::disposeMetadata, ioExecutor);

diff --git a/...luss/server/kv/snapshot/KvFileHandle.java → ...pache/fluss/kv/snapshot/KvFileHandle.java b/...luss/server/kv/snapshot/KvFileHandle.java → ...pache/fluss/kv/snapshot/KvFileHandle.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
+package org.apache.fluss.kv.snapshot;
 
 import org.apache.fluss.fs.FileSystem;
 import org.apache.fluss.fs.FsPath;

diff --git a/...kv/snapshot/KvFileHandleAndLocalPath.java → ...kv/snapshot/KvFileHandleAndLocalPath.java b/...kv/snapshot/KvFileHandleAndLocalPath.java → ...kv/snapshot/KvFileHandleAndLocalPath.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
+package org.apache.fluss.kv.snapshot;
 
 import java.util.Objects;
 

diff --git a/.../server/kv/snapshot/KvSnapshotHandle.java → ...e/fluss/kv/snapshot/KvSnapshotHandle.java b/.../server/kv/snapshot/KvSnapshotHandle.java → ...e/fluss/kv/snapshot/KvSnapshotHandle.java
@@ -15,9 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
-
-import org.apache.fluss.server.utils.SnapshotUtil;
+package org.apache.fluss.kv.snapshot;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,9 +24,6 @@
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-import static org.apache.fluss.utils.Preconditions.checkNotNull;
-import static org.apache.fluss.utils.Preconditions.checkState;
-
 /**
  * A handle to the snapshot of a kv tablet. It contains the share file handles and the private file
  * handles from which we can rebuild the kv tablet.
@@ -45,16 +40,6 @@ public class KvSnapshotHandle {
     /** The size of the incremental snapshot. */
     private final long incrementalSize;
 
-    /**
-     * Once the shared states are registered, it is the {@link SharedKvFileRegistry}'s
-     * responsibility to cleanup those shared states. But in the cases where the state handle is
-     * discarded before performing the registration, the handle should delete all the shared states
-     * created by it.
-     *
-     * <p>This variable is not null iff the handles was registered.
-     */
-    private transient SharedKvFileRegistry sharedKvFileRegistry;
-
     public KvSnapshotHandle(
             List<KvFileHandleAndLocalPath> sharedFileHandles,
             List<KvFileHandleAndLocalPath> privateFileHandles,
@@ -97,43 +82,35 @@ public long getSnapshotSize() {
         return snapshotSize;
     }
 
-    public void discard() {
-        SharedKvFileRegistry registry = this.sharedKvFileRegistry;
-        final boolean isRegistered = (registry != null);
-
+    /**
+     * Discards only the private files of this snapshot. Use this when shared files are managed
+     * externally by a registry.
+     */
+    public void discardPrivateFiles() {
         try {
             SnapshotUtil.bestEffortDiscardAllKvFiles(
                     privateFileHandles.stream()
                             .map(KvFileHandleAndLocalPath::getKvFileHandle)
                             .collect(Collectors.toList()));
         } catch (Exception e) {
-            LOG.warn("Could not properly discard misc file states.", e);
-        }
-
-        if (!isRegistered) {
-            try {
-                SnapshotUtil.bestEffortDiscardAllKvFiles(
-                        sharedFileHandles.stream()
-                                .map(KvFileHandleAndLocalPath::getKvFileHandle)
-                                .collect(Collectors.toSet()));
-            } catch (Exception e) {
-                LOG.warn("Could not properly discard new sst file states.", e);
-            }
+            LOG.warn("Could not properly discard private file states.", e);
         }
     }
 
-    public void registerKvFileHandles(SharedKvFileRegistry registry, long snapshotID) {
-        checkState(
-                sharedKvFileRegistry != registry,
-                "The kv file handle has already registered its shared kv files to the given registry.");
-
-        sharedKvFileRegistry = checkNotNull(registry);
+    /**
+     * Discards all files (private + shared). Use this for failed or aborted snapshots that were
+     * never registered with a shared file registry.
+     */
+    public void discardAll() {
+        discardPrivateFiles();
 
-        for (KvFileHandleAndLocalPath handleAndLocalPath : sharedFileHandles) {
-            registry.registerReference(
-                    SharedKvFileRegistryKey.fromKvFileHandle(handleAndLocalPath.getKvFileHandle()),
-                    handleAndLocalPath.getKvFileHandle(),
-                    snapshotID);
+        try {
+            SnapshotUtil.bestEffortDiscardAllKvFiles(
+                    sharedFileHandles.stream()
+                            .map(KvFileHandleAndLocalPath::getKvFileHandle)
+                            .collect(Collectors.toSet()));
+        } catch (Exception e) {
+            LOG.warn("Could not properly discard shared sst file states.", e);
         }
     }
 

diff --git a/...ache/fluss/server/utils/SnapshotUtil.java → ...pache/fluss/kv/snapshot/SnapshotUtil.java b/...ache/fluss/server/utils/SnapshotUtil.java → ...pache/fluss/kv/snapshot/SnapshotUtil.java
@@ -15,9 +15,8 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.utils;
+package org.apache.fluss.kv.snapshot;
 
-import org.apache.fluss.server.kv.snapshot.KvFileHandle;
 import org.apache.fluss.utils.ExceptionUtils;
 import org.apache.fluss.utils.function.ThrowingConsumer;
 
@@ -44,6 +43,7 @@ public static void bestEffortDiscardAllKvFiles(
         applyToAllWhileSuppressingExceptions(handlesToDiscard, KvFileHandle::discard);
     }
 
+    /** Quietly discards a single kv file handle, logging any exception. */
     public static void discardKvFileQuietly(KvFileHandle kvFileHandle) {
         if (kvFileHandle == null) {
             return;

diff --git a/.../snapshot/CompletedSnapshotJsonSerde.java → ...tils/json/CompletedSnapshotJsonSerde.java b/.../snapshot/CompletedSnapshotJsonSerde.java → ...tils/json/CompletedSnapshotJsonSerde.java
@@ -15,58 +15,78 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
+package org.apache.fluss.utils.json;
 
 import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.kv.autoinc.AutoIncIDRange;
+import org.apache.fluss.kv.snapshot.CompletedSnapshot;
+import org.apache.fluss.kv.snapshot.KvFileHandle;
+import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath;
+import org.apache.fluss.kv.snapshot.KvSnapshotHandle;
 import org.apache.fluss.metadata.TableBucket;
-import org.apache.fluss.server.kv.autoinc.AutoIncIDRange;
 import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
 import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
-import org.apache.fluss.utils.json.JsonDeserializer;
-import org.apache.fluss.utils.json.JsonSerdeUtils;
-import org.apache.fluss.utils.json.JsonSerializer;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
-/** Json serializer and deserializer for {@link CompletedSnapshot}. */
+/**
+ * Json serializer and deserializer for {@link CompletedSnapshot}.
+ *
+ * <p>This class is the single source of truth for KV snapshot {@code _METADATA} JSON format
+ * constants. It provides full serialization/deserialization as well as lightweight extraction
+ * utilities (e.g., {@link #parseSharedSstLocalPaths(byte[])}) for consumers that do not need a full
+ * domain object.
+ */
 public class CompletedSnapshotJsonSerde
         implements JsonSerializer<CompletedSnapshot>, JsonDeserializer<CompletedSnapshot> {
 
-    public static final CompletedSnapshotJsonSerde INSTANCE = new CompletedSnapshotJsonSerde();
-
-    private static final int VERSION = 1;
-    private static final String VERSION_KEY = "version";
-    // for table bucket the snapshot belongs to
-    private static final String TABLE_ID = "table_id";
-    private static final String PARTITION_ID = "partition_id";
-    private static final String BUCKET_ID = "bucket_id";
-
-    private static final String SNAPSHOT_ID = "snapshot_id";
-    private static final String SNAPSHOT_LOCATION = "snapshot_location";
-
-    // for kv snapshot's files
-    private static final String KV_SNAPSHOT_HANDLE = "kv_snapshot_handle";
-    private static final String KV_SHARED_FILES_HANDLE = "shared_file_handles";
-    private static final String KV_PRIVATE_FILES_HANDLE = "private_file_handles";
-    private static final String KV_FILE_HANDLE = "kv_file_handle";
-    private static final String KV_FILE_PATH = "path";
-    private static final String KV_FILE_SIZE = "size";
-    private static final String KV_FILE_LOCAL_PATH = "local_path";
-    private static final String SNAPSHOT_INCREMENTAL_SIZE = "snapshot_incremental_size";
-
     // ---------------------------------------------------------------------------------
-    // kv tablet state for the snapshot
+    // JSON field name constants — _METADATA schema version 1
     // ---------------------------------------------------------------------------------
 
-    // for the next log offset when the snapshot is triggered;
-    private static final String LOG_OFFSET = "log_offset";
-    private static final String ROW_COUNT = "row_count";
-    private static final String AUTO_INC_ID_RANGE = "auto_inc_id_range";
-    private static final String AUTO_INC_COLUMN_ID = "column_id";
-    private static final String AUTO_INC_ID_START = "start";
-    private static final String AUTO_INC_ID_END = "end";
+    /** Current schema version. */
+    public static final int VERSION = 1;
+
+    public static final String VERSION_KEY = "version";
+
+    // Table bucket identification
+    public static final String TABLE_ID = "table_id";
+    public static final String PARTITION_ID = "partition_id";
+    public static final String BUCKET_ID = "bucket_id";
+
+    // Snapshot identity
+    public static final String SNAPSHOT_ID = "snapshot_id";
+    public static final String SNAPSHOT_LOCATION = "snapshot_location";
+
+    // KV snapshot handle structure
+    public static final String KV_SNAPSHOT_HANDLE = "kv_snapshot_handle";
+    public static final String KV_SHARED_FILES_HANDLE = "shared_file_handles";
+    public static final String KV_PRIVATE_FILES_HANDLE = "private_file_handles";
+
+    // Individual file entry fields
+    public static final String KV_FILE_HANDLE = "kv_file_handle";
+    public static final String KV_FILE_PATH = "path";
+    public static final String KV_FILE_SIZE = "size";
+    public static final String KV_FILE_LOCAL_PATH = "local_path";
+
+    // Snapshot size
+    public static final String SNAPSHOT_INCREMENTAL_SIZE = "snapshot_incremental_size";
+
+    // KV tablet state
+    public static final String LOG_OFFSET = "log_offset";
+    public static final String ROW_COUNT = "row_count";
+
+    // Auto-increment ID ranges
+    public static final String AUTO_INC_ID_RANGE = "auto_inc_id_range";
+    public static final String AUTO_INC_COLUMN_ID = "column_id";
+    public static final String AUTO_INC_ID_START = "start";
+    public static final String AUTO_INC_ID_END = "end";
+
+    public static final CompletedSnapshotJsonSerde INSTANCE = new CompletedSnapshotJsonSerde();
 
     @Override
     public void serialize(CompletedSnapshot completedSnapshot, JsonGenerator generator)
@@ -240,6 +260,10 @@ private List<KvFileHandleAndLocalPath> deserializeKvFileHandles(
         return kvFileHandleAndLocalPaths;
     }
 
+    // ---------------------------------------------------------------------------------
+    // Static convenience methods
+    // ---------------------------------------------------------------------------------
+
     /** Serialize the {@link CompletedSnapshot} to json bytes. */
     public static byte[] toJson(CompletedSnapshot completedSnapshot) {
         return JsonSerdeUtils.writeValueAsBytes(completedSnapshot, INSTANCE);
@@ -249,4 +273,56 @@ public static byte[] toJson(CompletedSnapshot completedSnapshot) {
     public static CompletedSnapshot fromJson(byte[] json) {
         return JsonSerdeUtils.readValue(json, INSTANCE);
     }
+
+    /**
+     * Parses a {@code _METADATA} JSON payload and returns the set of shared SST file names (local
+     * path basenames) referenced by the snapshot.
+     *
+     * <p>This method navigates the JSON tree: {@code kv_snapshot_handle → shared_file_handles[*] →
+     * local_path} and collects non-empty values.
+     *
+     * @param metadataJsonBytes raw bytes of the {@code _METADATA} file
+     * @return set of shared SST file names (e.g. {@code "abc-def-0.sst"})
+     * @throws IOException if the bytes cannot be parsed as valid JSON or the expected structure is
+     *     missing
+     */
+    public static Set<String> parseSharedSstLocalPaths(byte[] metadataJsonBytes)
+            throws IOException {
+        JsonNode root = JsonSerdeUtils.OBJECT_MAPPER_INSTANCE.readTree(metadataJsonBytes);
+        JsonNode kvSnapshotHandle = root.get(KV_SNAPSHOT_HANDLE);
+        if (kvSnapshotHandle == null) {
+            throw new IOException("Missing '" + KV_SNAPSHOT_HANDLE + "' in _METADATA JSON payload");
+        }
+        JsonNode sharedFilesNode = kvSnapshotHandle.get(KV_SHARED_FILES_HANDLE);
+        if (sharedFilesNode == null || !sharedFilesNode.isArray()) {
+            throw new IOException(
+                    "Missing or non-array '"
+                            + KV_SHARED_FILES_HANDLE
+                            + "' in _METADATA JSON payload");
+        }
+
+        Set<String> fileNames = new HashSet<>();
+        for (JsonNode entry : sharedFilesNode) {
+            JsonNode localPathNode = entry.get(KV_FILE_LOCAL_PATH);
+            if (localPathNode == null || !localPathNode.isTextual()) {
+                throw new IOException(
+                        "Missing or non-textual '"
+                                + KV_FILE_LOCAL_PATH
+                                + "' in "
+                                + KV_SHARED_FILES_HANDLE
+                                + " entry");
+            }
+            String localPath = localPathNode.asText();
+            if (localPath.isEmpty()) {
+                throw new IOException(
+                        "Empty '"
+                                + KV_FILE_LOCAL_PATH
+                                + "' in "
+                                + KV_SHARED_FILES_HANDLE
+                                + " entry");
+            }
+            fileNames.add(localPath);
+        }
+        return fileNames;
+    }
 }