Resharding: expose types (#5718)

* Expose resharding related types * Still hide some fields, end users are not intended to use them * Remove unused JsonSchema derives
2026-08-06 18:10:58 -05:00 · 2025-01-14 09:15:11 +01:00
parent f02f319953
commit 2b7fdffb3a
5 changed files with 92 additions and 23 deletions
--- 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": [
--- 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<ShardId>,
    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<ShardId>,
    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<ShardId>,
    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<ShardId>,
    pub to_peer_id: PeerId,
    pub from_peer_id: PeerId,
--- 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<ShardId>,

    /// Source peer id
--- 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,
 }

--- 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<ShardId>,
    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<ShardId>,
    pub from: PeerId,
    pub to: PeerId,
@@ -92,11 +91,10 @@ impl From<ShardTransfer> 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<ShardId>,
    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,
 }