docs/design-docs/design_docs/20251114-snapshot_design.md
Milvus Snapshot mechanism provides complete collection-level data snapshot capabilities, implementing point-in-time backup and restore functionality. The implementation includes the following core components:
1. Snapshot Storage Layer (S3/Object Storage)
2. Snapshot Metadata Management (Etcd)
3. Collection Meta Restoration
4. Data Restore Mechanism (Copy Segment)
5. Garbage Collection Integration
Execution Steps:
MsgPosition for each collection channel. These positions form the snapshot's per-channel data boundary.create_ts = min(channel_seek_positions.timestamp) for legacy display and sorting.segmentEffectiveTs(segment) < channel_seek_positions[segment.channel_name].timestamp.Important Notes:
create_ts is a compatibility summary equal to the minimum channel seek timestamp. It is not a global cross-channel visibility boundary.Data Point-in-Time:
Adopts layered storage architecture with separation of responsibilities:
Etcd Storage Layer - Fast query and management
S3 Storage Layer - Complete data persistence
Data is organized using Iceberg-like manifest format with the following directory structure:
snapshots/{collection_id}/
├── metadata/
│ └── {snapshot_id}.json # Snapshot metadata (JSON format)
│
└── manifests/
└── {snapshot_id}/ # Directory for each snapshot
├── {segment_id_1}.avro # Individual segment manifest (Avro format)
├── {segment_id_2}.avro
└── ...
Design Note: Each segment has its own manifest file to enable:
1. metadata/{snapshot_id}.json (JSON format)
Structure of metadata JSON:
type SnapshotMetadata struct {
SnapshotInfo *datapb.SnapshotInfo // Snapshot basic info
Collection *datapb.CollectionDescription // Complete schema
Indexes []*indexpb.IndexInfo // Index definitions
ManifestList []string // Array of manifest file paths
SegmentIDs []int64 // Pre-computed segment IDs for fast reload
IndexIDs []int64 // Pre-computed index IDs for fast reload
StorageV2ManifestList []*StorageV2SegmentManifest // StorageV2 (Lance/Arrow) manifest mappings
}
type StorageV2SegmentManifest struct {
SegmentID int64 // Segment ID
Manifest string // Path to StorageV2 manifest file
}
Fast Reload Optimization: The SegmentIDs and IndexIDs fields enable DataCoord to quickly reload snapshot metadata during startup without parsing heavy Avro manifest files. Full segment data is loaded on-demand.
2. manifests/{snapshot_id}/{segment_id}.avro (Avro format)
StorageV2 Support: For segments using Milvus's newer storage format, additional manifest paths are stored in StorageV2ManifestList. This enables seamless snapshot/restore for both legacy and new storage formats.
1. Save() - Main entry point for writing snapshot
a. Create manifest directory: manifests/{snapshot_id}/
b. For each segment, call writeSegmentManifest():
- Convert SegmentDescription to Avro ManifestEntry
- Serialize to Avro binary format
- Write to S3: manifests/{snapshot_id}/{segment_id}.avro
c. Collect StorageV2 manifest paths (if applicable)
d. Call writeMetadataFile()
2. writeSegmentManifest() - Write individual segment manifest
- Convert SegmentDescription to ManifestEntry (Avro struct)
- Include all binlog, deltalog, statslog, index files
- Include position info (start_position, dml_position)
- Serialize and write to: manifests/{snapshot_id}/{segment_id}.avro
3. writeMetadataFile() - Write main metadata file
- Create SnapshotMetadata containing:
* SnapshotInfo
* Collection schema
* Index definitions
* ManifestList array (paths from step 1)
* SegmentIDs/IndexIDs (pre-computed for fast reload)
* StorageV2ManifestList (if applicable)
- Serialize to JSON format
- Write to S3: metadata/{snapshot_id}.json
1. ReadSnapshot(metadataFilePath, includeSegments) - Read snapshot data
a. Read and parse metadata JSON file
b. If includeSegments=false:
- Return only SnapshotInfo, Collection, Indexes
- Use pre-computed SegmentIDs/IndexIDs for fast reload
c. If includeSegments=true:
- For each path in ManifestList array:
* Read Avro binary data
* Parse ManifestEntry records
* Convert to SegmentDescription
- Populate StorageV2 manifest paths if present
d. Return complete SnapshotData
2. ListSnapshots(collectionID) - List all snapshots for a collection
- List all files in metadata/ directory
- Parse each metadata file to extract SnapshotInfo
- Return list of SnapshotInfo
Performance Optimization: The includeSegments parameter allows DataCoord to defer loading heavy segment data. During startup, only metadata is loaded (includeSegments=false), and full segment data is loaded on-demand when needed for restore operations.
SnapshotManager centralizes all snapshot-related business logic, providing a unified interface for snapshot lifecycle management and restore operations.
Design Principles:
Core Interface:
type SnapshotManager interface {
// Snapshot lifecycle management
CreateSnapshot(ctx context.Context, collectionID int64, name, description string) (int64, error)
DropSnapshot(ctx context.Context, name string) error
DescribeSnapshot(ctx context.Context, name string) (*SnapshotData, error)
ListSnapshots(ctx context.Context, collectionID, partitionID int64) ([]string, error)
// Restore operations
RestoreSnapshot(ctx context.Context, snapshotName string, targetCollectionID int64) (int64, error)
GetRestoreState(ctx context.Context, jobID int64) (*datapb.RestoreSnapshotInfo, error)
ListRestoreJobs(ctx context.Context, collectionID int64) ([]*datapb.RestoreSnapshotInfo, error)
}
SnapshotMeta manages snapshot reference relationships and lifecycle:
type SnapshotDataInfo struct {
snapshotInfo *datapb.SnapshotInfo
SegmentIDs typeutil.UniqueSet // List of segments referenced by this snapshot
IndexIDs typeutil.UniqueSet // List of indexes referenced by this snapshot
}
type snapshotMeta struct {
catalog metastore.DataCoordCatalog
snapshotID2DataInfo *typeutil.ConcurrentMap[UniqueID, *SnapshotDataInfo]
reader *SnapshotReader
writer *SnapshotWriter
}
Core Functions:
1. SaveSnapshot() - Two-Phase Commit (2PC)
SaveSnapshot uses a two-phase commit approach to ensure atomic creation and enable GC cleanup of orphan files:
Phase 1 (Prepare):
- Save snapshot with PENDING state to catalog (etcd)
- Snapshot ID is allocated and stored
- PendingStartTime is recorded for timeout tracking
Write S3 Data:
- Write segment manifest files to S3
- Write metadata.json to S3
- Update S3Location in snapshot info
Phase 2 (Commit):
- Update snapshot state to COMMITTED in catalog
- Insert into in-memory cache with precomputed ID sets
Failure Handling:
| Failure Point | State | Recovery Action |
|---|---|---|
| Phase 1 fails | No changes | Return error, no cleanup needed |
| S3 write fails | PENDING in catalog | GC will cleanup S3 files using snapshot ID |
| Phase 2 fails | PENDING in catalog, S3 data exists | GC will cleanup S3 files and catalog record |
Key Design Points:
PendingStartTime + timeout to identify orphaned snapshots2. DropSnapshot()
3. GetSnapshotBySegment()/GetSnapshotByIndex()
4. GetPendingSnapshots()
5. CleanupPendingSnapshot()
Pending Snapshot GC (recyclePendingSnapshots):
Cleans up orphaned snapshot files from failed 2PC commits:
Process flow:
1. Get all PENDING snapshots from catalog that have exceeded timeout
2. For each pending snapshot:
a. Compute manifest directory: snapshots/{collection_id}/manifests/{snapshot_id}/
b. Compute metadata file: snapshots/{collection_id}/metadata/{snapshot_id}.json
c. Delete manifest directory using RemoveWithPrefix (no S3 list needed)
d. Delete metadata file
e. Delete etcd record
Key Design Points:
SnapshotPendingTimeout) prevents cleanup of snapshots still being createdConfiguration:
dataCoord:
gc:
snapshotPendingTimeout: 10m # Time before pending snapshot is considered orphaned
Segment GC Protection:
Snapshot protection mechanism is integrated in DataCoord's garbage_collector:
func (gc *garbageCollector) isSnapshotSegment(collectionID, segmentID int64) bool {
snapshotIDs := gc.meta.GetSnapshotMeta().GetSnapshotBySegment(ctx, collectionID, segmentID)
return len(snapshotIDs) > 0
}
Index GC Protection:
Index protection mechanism is integrated in IndexCoord:
func (gc *garbageCollector) recycleUnusedIndexes() {
// Check if index is referenced by snapshots
snapshotIDs := gc.meta.GetSnapshotMeta().GetSnapshotByIndex(ctx, collectionID, indexID)
if len(snapshotIDs) > 0 {
// Skip index files that are still referenced by snapshots
continue
}
}
Create:
User Request -> Proxy CreateSnapshot -> DataCoord
-> GetSnapshotSeekPositions() (obtain one seek position per channel)
-> compute create_ts as min(channel_seek_positions.timestamp)
-> SelectSegments() (filter each segment by its own channel seek timestamp)
-> Collect Schema/Indexes/Segment detailed information
-> SnapshotWriter.Save() (write to S3)
-> SnapshotMeta.SaveSnapshot() (write to Etcd + memory)
Drop:
User Request -> Proxy DropSnapshot -> DataCoord
-> SnapshotMeta.DropSnapshot()
-> Remove from Etcd
-> SnapshotWriter.Drop() (delete S3 files)
-> Remove from memory
List:
User Request -> Proxy ListSnapshots -> DataCoord
-> SnapshotMeta.ListSnapshots()
-> Filter by collection/partition
Describe:
User Request -> Proxy DescribeSnapshot -> DataCoord
-> SnapshotMeta.GetSnapshot()
-> Return SnapshotInfo
Restore:
User Request -> Proxy RestoreSnapshot -> RootCoord
-> DescribeSnapshot() from DataCoord (get schema, partitions, indexes)
-> CreateCollection() with PreserveFieldId=true
-> CreatePartition() for each user partition
-> RestoreSnapshotDataAndIndex() to DataCoord
-> Create indexes with PreserveIndexId=true
-> Create CopySegmentJob
-> Return job_id for async tracking
Restore is implemented using the Copy Segment mechanism, which significantly improves recovery speed by directly copying segment files, avoiding the overhead of rewriting data and rebuilding indexes.
The restore operation follows a layered architecture with clear separation of responsibilities:
User Request
│
▼
┌─────────┐
│ Proxy │ ← Entry point, request validation
└────┬────┘
│
▼
┌───────────┐
│ RootCoord │ ← Orchestration: CreateCollection, CreatePartition, coordinate DataCoord
└─────┬─────┘
│
▼
┌───────────┐
│ DataCoord │ ← Data restoration: CopySegmentJob creation, index creation
└─────┬─────┘
│
▼
┌──────────┐
│ DataNode │ ← Execution: Copy segment files from S3
└──────────┘
1. Proxy RestoreSnapshotTask.Execute():
- Validates request parameters
- Delegates to RootCoord.RestoreSnapshot()
- RootCoord orchestrates the entire restore process
Design Notes:
2. RootCoord restoreSnapshotTask.Execute():
a. Get snapshot info from DataCoord
- DescribeSnapshot() retrieves complete snapshot information (schema, partitions, indexes)
b. Create collection with preserved field IDs
- CreateCollection() with PreserveFieldId=true
- Schema properties are completely copied (consistency level, num_shards, etc.)
c. Create user partitions
- CreatePartition() for each user-created partition in snapshot
d. Restore data and indexes via DataCoord
- RestoreSnapshotDataAndIndex() handles data copying and index creation
Key Design Points:
PreserveFieldId=true: Ensures new collection field IDs match those in snapshot3. DataCoord RestoreSnapshotDataAndIndex():
a. Read snapshot data
- SnapshotMeta.ReadSnapshotData() reads complete segment information from S3
b. Generate mappings
- Channel Mapping: Map snapshot channels to new collection channels
- Partition Mapping: Map snapshot partition IDs to new partition IDs
c. Create indexes (if requested)
- Create indexes with PreserveIndexId=true
- Ensures index IDs match, allowing direct use of index files from snapshot
d. Pre-register Target Segments
- Allocate new segment IDs
- Create SegmentInfo (with correct PartitionID and InsertChannel)
- Pre-register to meta via meta.AddSegment(), State set to Importing
e. Create Copy Segment Job
- Create lightweight ID mappings (source_segment_id -> target_segment_id)
- Generate CopySegmentJob and save to meta
- CopySegmentChecker automatically creates CopySegmentTasks
ID Mapping Structure:
message CopySegmentIDMapping {
int64 source_segment_id = 1; // Segment ID in snapshot
int64 target_segment_id = 2; // Segment ID in new collection
int64 partition_id = 3; // Target partition ID (cached for grouping)
}
Design Notes:
meta.AddSegment()PartitionID, InsertChannel, State, etc.4. CopySegmentChecker monitors job execution:
Pending -> Executing:
- Group segments by maxSegmentsPerCopyTask (currently default 10/group)
- Create one CopySegmentTask for each group
- Distribute tasks to DataNodes for execution
Executing:
- DataNode CopySegmentTask executes:
a. Read all file paths of source segment
b. Copy files to new paths (update segment_id/partition_id/channel_name)
c. Update segment's binlog/deltalog/indexfile information in meta
- CopySegmentChecker monitors progress of all tasks
Executing -> Completed:
- After all tasks complete
- Update all target segments status to Flushed
- Job marked as Completed
Task Granularity Concurrency Design:
Create tasks grouped by maxSegmentsPerCopyTask (default 10 segments)
Each group generates independent CopySegmentTask, which can be distributed to different DataNodes for parallel execution
Long-term Optimization: If copy performance is insufficient, implement concurrency acceleration at task level
Copy Segment Details:
File copy operations on DataNode:
// For each segment in the task
1. Copy Binlog files
- Read all binlog files of source segment
- Update segment_id/partition_id/channel in file paths
- Copy to new paths
2. Copy Deltalog files
- Process delete logs
- Update paths and copy
3. Copy Statslog files
- Process statistics files
- Update paths and copy
4. Copy Index files
- Copy vector index, scalar index
- Keep index ID unchanged (because PreserveIndexId=true)
5. Update Segment metadata
- Create new SegmentInfo with target_segment_id
- Preserve num_of_rows, storage_version, and other information
- State set to Importing (will be updated to Flushed later)
CopySegmentJob:
message CopySegmentJob {
int64 job_id = 1;
int64 db_id = 2;
int64 collection_id = 3;
string collection_name = 4;
CopySegmentJobState state = 5;
string reason = 6;
repeated CopySegmentIDMapping id_mappings = 7; // Lightweight ID mapping (~48 bytes per segment)
uint64 timeout_ts = 8;
uint64 cleanup_ts = 9;
string start_time = 10;
string complete_time = 11;
repeated common.KeyValuePair options = 12; // Option configuration (e.g. copy_index)
int64 total_segments = 13;
int64 copied_segments = 14;
int64 total_rows = 15;
string snapshot_name = 16; // For restore snapshot scenario
}
CopySegmentTask:
message CopySegmentTask {
int64 task_id = 1;
int64 job_id = 2;
int64 collection_id = 3;
int64 node_id = 4; // DataNode ID executing the task
int64 task_version = 5;
int64 task_slot = 6;
ImportTaskStateV2 state = 7;
string reason = 8;
repeated CopySegmentIDMapping id_mappings = 9; // Segments this task is responsible for
string created_time = 10;
string complete_time = 11;
}
Task Scheduling:
DataCoord Layer Automatic Retry:
Pending state and wait for reschedulingRetry status, reset task to Pending stateRetry Configuration:
max_retries: Default 3 timesretry_count: Records current retry countRetry Flow:
PendingCopySegmentInspector periodically scans tasks in Pending stateDataNode Layer Failure Strategy:
Failed status1. Task-level Cleanup (Inspector):
CopySegmentInspector automatically iterates through all ID mappings of the taskDropped2. Job-level Cleanup (Checker):
CopySegmentChecker finds all tasks in Pending and InProgress states for that jobFailed status and synchronizes failure reason3. Timeout Mechanism:
CopySegmentChecker periodically checks job's timeout_tsFailed4. GC Mechanism:
CopySegmentChecker regularly checks jobs in Completed or Failed statecleanup_ts time is reached, sequentially deletes tasks and job metadataGC Protection Strategy:
Completed or Failed stateCleanupTs (based on ImportTaskRetention configuration)| Mechanism | Trigger Condition | Action | Purpose |
|---|---|---|---|
| Auto Retry | Query failure, DataNode returns Retry | Reset to Pending, reschedule | Automatically recover temporary errors |
| Task Cleanup | Task marked as Failed | Delete target segments | Clean up invalid meta data |
| Job Cleanup | Job marked as Failed | Cascade mark all tasks | Ensure state consistency |
| Timeout Protection | Exceeds TimeoutTs | Mark job as Failed | Prevent long-term resource occupation |
| GC Collection | Reaches CleanupTs | Delete job/task metadata | Prevent meta bloat |
State Machine:
RestoreSnapshotPending
↓
RestoreSnapshotExecuting (copying segments)
↓
RestoreSnapshotCompleted / RestoreSnapshotFailed
Progress Calculation:
Progress = (copied_segments / total_segments) * 100
Job Monitoring APIs:
Advantages:
Comparison with Bulk Insert:
1. Collection Creation Failure:
2. Copy Task Failure:
3. Job Timeout and GC:
1. Channel/Partition Count Must Match:
2. Field ID and Index ID Preservation:
3. TTL Handling:
Reading snapshot data through Milvus service:
Offline access through snapshot data on S3:
Core Design Goals:
Technical Implementation:
Function: Create snapshot for specified collection, automatically export data to object storage
Request:
message CreateSnapshotRequest {
common.MsgBase base = 1;
string db_name = 2; // database name
string collection_name = 3; // collection name
string name = 4; // user-defined snapshot name (unique)
string description = 5; // user-defined snapshot description
}
Response:
message CreateSnapshotResponse {
common.Status status = 1;
}
Implementation Details:
create_ts.Notes:
Function: Delete specified snapshot and all its files on S3
Request:
message DropSnapshotRequest {
common.MsgBase base = 1;
string name = 2; // snapshot name to drop
}
Response:
message DropSnapshotResponse {
common.Status status = 1;
}
Implementation Details:
Function: List all snapshots for specified collection
Request:
message ListSnapshotsRequest {
common.MsgBase base = 1;
string db_name = 2; // database name
string collection_name = 3; // collection name
}
Response:
message ListSnapshotsResponse {
common.Status status = 1;
repeated string snapshots = 2; // list of snapshot names
}
Function: Get detailed information about snapshot
Request:
message DescribeSnapshotRequest {
common.MsgBase base = 1;
string name = 2; // snapshot name
}
Response:
message DescribeSnapshotResponse {
common.Status status = 1;
string name = 2;
string description = 3;
int64 create_ts = 4;
string collection_name = 5;
repeated string partition_names = 6;
}
Return Information:
Function: Restore data from snapshot to new collection
Request (milvuspb - User-facing API):
message RestoreSnapshotRequest {
common.MsgBase base = 1;
string name = 2; // snapshot name to restore
string db_name = 3; // target database name
string collection_name = 4; // target collection name (must not exist)
}
Response:
message RestoreSnapshotResponse {
common.Status status = 1;
int64 job_id = 2; // restore job ID for progress tracking
}
Internal API (datapb - DataCoord API):
// RestoreSnapshotDataAndIndex - Called by RootCoord after collection/partition creation
message RestoreSnapshotRequest {
common.MsgBase base = 1;
string name = 2; // snapshot name
int64 collection_id = 3; // target collection ID (already created by RootCoord)
bool create_indexes = 4; // whether to create indexes during restore
}
Implementation Flow:
Limitations:
Function: Query restore job execution status
Request:
message GetRestoreSnapshotStateRequest {
common.MsgBase base = 1;
int64 job_id = 2; // restore job ID
}
Response:
message GetRestoreSnapshotStateResponse {
common.Status status = 1;
RestoreSnapshotInfo info = 2;
}
message RestoreSnapshotInfo {
int64 job_id = 1;
string snapshot_name = 2;
string db_name = 3;
string collection_name = 4;
RestoreSnapshotState state = 5; // Pending/Executing/Completed/Failed
int32 progress = 6; // 0-100
string reason = 7; // error reason if failed
uint64 time_cost = 8; // milliseconds
}
State Values:
Function: List all restore jobs (optionally filter by collection)
Request:
message ListRestoreSnapshotJobsRequest {
common.MsgBase base = 1;
string collection_name = 2; // optional: filter by collection
}
Response:
message ListRestoreSnapshotJobsResponse {
common.Status status = 1;
repeated RestoreSnapshotInfo jobs = 2;
}