diff --git a/docs/redoc/master/openapi.json b/docs/redoc/master/openapi.json index 821b8c5395..c9db70f746 100644 --- a/docs/redoc/master/openapi.json +++ b/docs/redoc/master/openapi.json @@ -10711,7 +10711,9 @@ "Initializing", "Listener", "PartialSnapshot", - "Recovery" + "Recovery", + "Resharding", + "ReshardingScaleDown" ] }, "RemoteShardInfo": { @@ -10764,6 +10766,13 @@ "format": "uint32", "minimum": 0 }, + "to_shard_id": { + "description": "Target shard ID if different than source shard ID\n\nUsed exclusively with `ReshardStreamRecords` transfer method.", + "type": "integer", + "format": "uint32", + "minimum": 0, + "nullable": true + }, "from": { "description": "Source peer id", "type": "integer", @@ -10820,6 +10829,13 @@ "enum": [ "wal_delta" ] + }, + { + "description": "Shard transfer for resharding: stream all records in batches until all points are transferred.", + "type": "string", + "enum": [ + "resharding_stream_records" + ] } ] }, @@ -12401,6 +12417,12 @@ }, { "$ref": "#/components/schemas/RestartTransferOperation" + }, + { + "$ref": "#/components/schemas/StartReshardingOperation" + }, + { + "$ref": "#/components/schemas/AbortReshardingOperation" } ] }, @@ -12672,6 +12694,58 @@ } } }, + "StartReshardingOperation": { + "type": "object", + "required": [ + "start_resharding" + ], + "properties": { + "start_resharding": { + "$ref": "#/components/schemas/StartResharding" + } + } + }, + "StartResharding": { + "type": "object", + "required": [ + "direction" + ], + "properties": { + "direction": { + "$ref": "#/components/schemas/ReshardingDirection" + }, + "peer_id": { + "type": "integer", + "format": "uint64", + "minimum": 0, + "nullable": true + }, + "shard_key": { + "anyOf": [ + { + "$ref": "#/components/schemas/ShardKey" + }, + { + "nullable": true + } + ] + } + } + }, + "AbortReshardingOperation": { + "type": "object", + "required": [ + "abort_resharding" + ], + "properties": { + "abort_resharding": { + "$ref": "#/components/schemas/AbortResharding" + } + } + }, + "AbortResharding": { + "type": "object" + }, "SearchRequestBatch": { "type": "object", "required": [ diff --git a/lib/collection/src/operations/cluster_ops.rs b/lib/collection/src/operations/cluster_ops.rs index c3854cf480..369aed4ec6 100644 --- a/lib/collection/src/operations/cluster_ops.rs +++ b/lib/collection/src/operations/cluster_ops.rs @@ -29,22 +29,20 @@ pub enum ClusterOperations { RestartTransfer(RestartTransferOperation), /// Start resharding - #[schemars(skip)] StartResharding(StartReshardingOperation), /// Finish migrating points on specified shard, mark shard as `Active` - #[schemars(skip)] + #[schemars(skip)] // hide for internal use FinishMigratingPoints(FinishMigratingPointsOperation), /// Commit read hashring - #[schemars(skip)] + #[schemars(skip)] // hide for internal use CommitReadHashRing(CommitReadHashRingOperation), /// Commit write hashring - #[schemars(skip)] + #[schemars(skip)] // hide for internal use CommitWriteHashRing(CommitWriteHashRingOperation), /// Finish resharding - #[schemars(skip)] + #[schemars(skip)] // hide for internal use FinishResharding(FinishReshardingOperation), /// Abort resharding - #[schemars(skip)] AbortResharding(AbortReshardingOperation), } @@ -93,7 +91,7 @@ pub struct DropShardingKey { pub struct RestartTransfer { pub shard_id: ShardId, #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding + #[schemars(skip)] // hide for internal use pub to_shard_id: Option, pub from_peer_id: PeerId, pub to_peer_id: PeerId, @@ -191,7 +189,7 @@ pub struct FinishReshardingOperation { pub struct ReplicateShard { pub shard_id: ShardId, #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding + #[schemars(skip)] // hide for internal use pub to_shard_id: Option, pub to_peer_id: PeerId, pub from_peer_id: PeerId, @@ -215,7 +213,7 @@ impl Validate for ReplicateShard { pub struct MoveShard { pub shard_id: ShardId, #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding + #[schemars(skip)] // hide for internal use pub to_shard_id: Option, pub to_peer_id: PeerId, pub from_peer_id: PeerId, @@ -257,7 +255,7 @@ pub struct Replica { pub struct AbortShardTransfer { pub shard_id: ShardId, #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding + #[schemars(skip)] // hide for internal use pub to_shard_id: Option, pub to_peer_id: PeerId, pub from_peer_id: PeerId, diff --git a/lib/collection/src/operations/types.rs b/lib/collection/src/operations/types.rs index fdd9b728ef..2c641c51fd 100644 --- a/lib/collection/src/operations/types.rs +++ b/lib/collection/src/operations/types.rs @@ -340,8 +340,10 @@ pub struct CollectionClusterInfo { pub struct ShardTransferInfo { pub shard_id: ShardId, + /// Target shard ID if different than source shard ID + /// + /// Used exclusively with `ReshardStreamRecords` transfer method. #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding pub to_shard_id: Option, /// Source peer id diff --git a/lib/collection/src/shards/replica_set/mod.rs b/lib/collection/src/shards/replica_set/mod.rs index 0cfbbf6475..c687459d14 100644 --- a/lib/collection/src/shards/replica_set/mod.rs +++ b/lib/collection/src/shards/replica_set/mod.rs @@ -1220,10 +1220,8 @@ pub enum ReplicaState { // Normally rejects updates, accepts updates if force is true Recovery, // Points are being migrated to this shard as part of resharding up - #[schemars(skip)] Resharding, // Points are being migrated to this shard as part of resharding down - #[schemars(skip)] ReshardingScaleDown, } diff --git a/lib/collection/src/shards/transfer/mod.rs b/lib/collection/src/shards/transfer/mod.rs index 8232b0735f..bc1c392f70 100644 --- a/lib/collection/src/shards/transfer/mod.rs +++ b/lib/collection/src/shards/transfer/mod.rs @@ -28,13 +28,13 @@ const CONSENSUS_CONFIRM_RETRY_DELAY: Duration = Duration::from_secs(1); /// Time after which confirming a consensus operation times out. const CONSENSUS_CONFIRM_TIMEOUT: Duration = defaults::CONSENSUS_META_OP_WAIT; -#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)] pub struct ShardTransfer { pub shard_id: ShardId, - /// For resharding, a different target shard ID may be configured - /// By default the shard ID on the target peer is the same. + /// Target shard ID if different than source shard ID + /// + /// Used exclusively with `ReshardStreamRecords` transfer method. #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding pub to_shard_id: Option, pub from: PeerId, pub to: PeerId, @@ -57,11 +57,10 @@ impl ShardTransfer { } } -#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)] pub struct ShardTransferRestart { pub shard_id: ShardId, #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding pub to_shard_id: Option, pub from: PeerId, pub to: PeerId, @@ -92,11 +91,10 @@ impl From for ShardTransferRestart { } /// Unique identifier of a transfer, agnostic of transfer method -#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)] pub struct ShardTransferKey { pub shard_id: ShardId, #[serde(default, skip_serializing_if = "Option::is_none")] - #[schemars(skip)] // TODO(resharding): expose once we release resharding pub to_shard_id: Option, pub from: PeerId, pub to: PeerId, @@ -121,7 +119,6 @@ pub enum ShardTransferMethod { WalDelta, /// Shard transfer for resharding: stream all records in batches until all points are /// transferred. - #[schemars(skip)] ReshardingStreamRecords, }