nafraf · Feb 16, 2026 · Jan 27, 2026 · Jan 30, 2026 · Jan 31, 2026 · Jan 31, 2026
diff --git a/commands.json b/commands.json
@@ -2094,6 +2094,13 @@
                     "token": "EF_RUNTIME",
                     "optional": true
                   },
+                  {
+                    "name": "shard_k_ratio",
+                    "type": "double",
+                    "token": "SHARD_K_RATIO",
+                    "optional": true,
+                    "since": "8.6.1"
+                  },
                   {
                     "name": "yield_score_as",
                     "type": "string",

diff --git a/src/command_info/command_info.c b/src/command_info/command_info.c
@@ -2347,6 +2347,13 @@ int SetFtHybridInfo(RedisModuleCommand *cmd) {
                     .type = REDISMODULE_ARG_TYPE_INTEGER,
                     .flags = REDISMODULE_CMD_ARG_OPTIONAL,
                   },
+                  {
+                    .name = "shard_k_ratio",
+                    .token = "SHARD_K_RATIO",
+                    .since = "8.6.1",
+                    .type = REDISMODULE_ARG_TYPE_DOUBLE,
+                    .flags = REDISMODULE_CMD_ARG_OPTIONAL,
+                  },
                   {
                     .name = "yield_score_as",
                     .token = "YIELD_SCORE_AS",

diff --git a/src/concurrent_ctx.c b/src/concurrent_ctx.c
@@ -48,6 +48,7 @@ typedef struct ConcurrentCmdCtx {
   int options;
   WeakRef spec_ref;
   rs_wall_clock_ns_t coordStartTime;  // Time when command was received on coordinator
+  size_t numShards;                   // Number of shards in the cluster (captured from main thread)
 } ConcurrentCmdCtx;
 
 /* Run a function on the concurrent thread pool */
@@ -99,6 +100,10 @@ rs_wall_clock_ns_t ConcurrentCmdCtx_GetCoordStartTime(ConcurrentCmdCtx *cctx) {
   return cctx->coordStartTime;
 }
 
+size_t ConcurrentCmdCtx_GetNumShards(const ConcurrentCmdCtx *cctx) {
+  return cctx->numShards;
+}
+
 int ConcurrentSearch_HandleRedisCommandEx(int poolType, ConcurrentCmdHandler handler,
                                           RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
                                           ConcurrentSearchHandlerCtx *handlerCtx) {
@@ -108,6 +113,7 @@ int ConcurrentSearch_HandleRedisCommandEx(int poolType, ConcurrentCmdHandler han
   cmdCtx->argc = argc;
   cmdCtx->spec_ref = handlerCtx->spec_ref;
   cmdCtx->coordStartTime = handlerCtx->coordStartTime;
+  cmdCtx->numShards = handlerCtx->numShards;
   cmdCtx->ctx = RedisModule_GetThreadSafeContext(cmdCtx->bc);
   RS_AutoMemory(cmdCtx->ctx);
   cmdCtx->handler = handler;

diff --git a/src/concurrent_ctx.h b/src/concurrent_ctx.h
@@ -50,6 +50,7 @@ typedef struct ConcurrentSearchHandlerCtx {
   rs_wall_clock_ns_t coordQueueTime;  // Time spent waiting in coordinator thread pool queue
   WeakRef spec_ref;                   // Weak reference to the index spec
   bool isProfile;                     // Whether this is an FT.PROFILE command
+  size_t numShards;                   // Number of shards in the cluster (captured from main thread)
 } ConcurrentSearchHandlerCtx;
 
 #define CMDCTX_KEEP_RCTX 0x01
@@ -73,6 +74,9 @@ WeakRef ConcurrentCmdCtx_GetWeakRef(struct ConcurrentCmdCtx *cctx);
 // Returns the coordinator start time held in the context.
 rs_wall_clock_ns_t ConcurrentCmdCtx_GetCoordStartTime(struct ConcurrentCmdCtx *cctx);
 
+// Returns the number of shards captured from the main thread.
+size_t ConcurrentCmdCtx_GetNumShards(const struct ConcurrentCmdCtx *cctx);
+
 /* Same as handleRedis command, but set flags for the concurrent context */
 int ConcurrentSearch_HandleRedisCommandEx(int poolType, ConcurrentCmdHandler handler,
                                           RedisModuleCtx *ctx, RedisModuleString **argv, int argc,

diff --git a/src/coord/hybrid/dist_hybrid.c b/src/coord/hybrid/dist_hybrid.c
@@ -21,6 +21,8 @@
 #include "info/global_stats.h"
 #include "profile/profile.h"
 #include "dist_profile.h"
+#include "shard_window_ratio.h"
+#include "config.h"
 
 // We mainly need the resp protocol to be three in order to easily extract the "score" key from the response
 #define HYBRID_RESP_PROTOCOL_VERSION 3
@@ -111,14 +113,26 @@ static int MRCommand_appendVsimFilter(MRCommand *xcmd, RedisModuleString **argv,
 
 /**
  * Appends all VSIM-related arguments to MR command.
- * This includes VSIM keyword, field, vector, KNN/RANGE method, and VSIM FILTER if present.
+ * This includes VSIM keyword, field, vector, KNN/RANGE method, and VSIM FILTER
+ * if present.
+ *
+ * SHARD_K_RATIO is only valid for KNN queries, but this is validated during
+ * parsing - no validation is done here.
  *
  * @param xcmd - destination MR command to append arguments to
  * @param argv - source command arguments array
  * @param argc - total argument count
  * @param vsimOffset - offset where VSIM keyword appears
+ * @param kArgIndex - output parameter for the index of the K value argument in
+ *        the built command (set to -1 if no KNN K argument found). Can be NULL.
  */
-static void MRCommand_appendVsim(MRCommand *xcmd, RedisModuleString **argv, int argc, int vsimOffset) {
+static void MRCommand_appendVsim(MRCommand *xcmd, RedisModuleString **argv,
+                                 int argc, int vsimOffset, int *kArgIndex) {
+  RS_LOG_ASSERT(kArgIndex, "kArgIndex must not be NULL");
+
+  // Initialize output parameter
+  *kArgIndex = -1;
+
   // Add VSIM keyword and field
   MRCommand_AppendRstr(xcmd, argv[vsimOffset]);     // VSIM
   MRCommand_AppendRstr(xcmd, argv[vsimOffset + 1]); // field
@@ -148,9 +162,27 @@ static void MRCommand_appendVsim(MRCommand *xcmd, RedisModuleString **argv, int
     if (vectorMethodOffset < argc - 2) {
       RedisModule_StringToLongLong(argv[vectorMethodOffset + 1], &methodNargs);
 
-      // Append method name, argument count, and all method arguments
-      for (int i = 0; i < methodNargs + 2; ++i) {
-        MRCommand_AppendRstr(xcmd, argv[vectorMethodOffset + i]);
+      // Append method name (KNN/RANGE) and argument count
+      MRCommand_AppendRstr(xcmd, argv[vectorMethodOffset]);     // KNN or RANGE
+      MRCommand_AppendRstr(xcmd, argv[vectorMethodOffset + 1]); // argument count
+
+      // Append method arguments
+      // Format for KNN: KNN <count> K <value> [EF_RUNTIME <value>]...
+      for (int i = 2; i < methodNargs + 2; ++i) {
+        size_t argLen;
+        const char *argStr = RedisModule_StringPtrLen(argv[vectorMethodOffset + i], &argLen);
+        bool kFound = (argLen == 1 && strncasecmp(argStr, "K", 1) == 0);
+
+        if (*kArgIndex == -1 && kFound && i + 1 < methodNargs + 2) {
+          // Found K keyword - append it and record position of the K value
+          MRCommand_AppendRstr(xcmd, argv[vectorMethodOffset + i]);  // K keyword
+          ++i;  // Move to K value
+          *kArgIndex = xcmd->num;  // Record position where K value will be appended
+          MRCommand_AppendRstr(xcmd, argv[vectorMethodOffset + i]);  // K value
+        } else {
+          // Regular argument - append as-is
+          MRCommand_AppendRstr(xcmd, argv[vectorMethodOffset + i]);
+        }
       }
     }
   }
@@ -193,7 +225,8 @@ static void MRCommand_appendVsim(MRCommand *xcmd, RedisModuleString **argv, int
 void HybridRequest_buildMRCommand(RedisModuleString **argv, int argc,
                             ProfileOptions profileOptions,
                             MRCommand *xcmd, arrayof(char*) serialized,
-                            IndexSpec *sp, HybridPipelineParams *hybridParams) {
+                            IndexSpec *sp, const VectorQuery *vq,
+                            size_t numShards) {
   int argOffset;
   const char *index_name = RedisModule_StringPtrLen(argv[1], NULL);
 
@@ -216,7 +249,17 @@ void HybridRequest_buildMRCommand(RedisModuleString **argv, int argc,
 
   // Add all VSIM-related arguments (VSIM, field, vector, methods, filter)
   int vsimOffset = RMUtil_ArgIndex("VSIM", argv, argc);
-  MRCommand_appendVsim(xcmd, argv, argc, vsimOffset);
+  int kArgIndex = -1;
+  MRCommand_appendVsim(xcmd, argv, argc, vsimOffset, &kArgIndex);
+
+  // Calculate and apply effective K for KNN queries if SHARD_K_RATIO is set
+  if (vq && vq->type == VECSIM_QT_KNN) {
+    double shardWindowRatio = vq->knn.shardWindowRatio;
+    if (shardWindowRatio < MAX_SHARD_WINDOW_RATIO && numShards > 1) {
+      size_t effectiveK = calculateEffectiveK(vq->knn.k, shardWindowRatio, numShards);
+      modifyVsimKNN(xcmd, kArgIndex, effectiveK, vq->knn.k);
+    }
+  }
 
   int combineOffset = RMUtil_ArgIndex("COMBINE", argv + vsimOffset, argc - vsimOffset);
   combineOffset = combineOffset != -1 ? combineOffset + vsimOffset : -1;
@@ -526,8 +569,9 @@ void printDistHybridProfile(RedisModule_Reply *reply, void *ctx) {
                         printDistHybridCoordinatorProfile, ctx);
 }
 
-static int HybridRequest_prepareForExecution(HybridRequest *hreq, RedisModuleCtx *ctx,
-        RedisModuleString **argv, int argc, IndexSpec *sp, QueryError *status) {
+static int HybridRequest_prepareForExecution(HybridRequest *hreq,
+        RedisModuleCtx *ctx, RedisModuleString **argv, int argc, IndexSpec *sp,
+        size_t numShards, QueryError *status) {
 
     hreq->tailPipeline->qctx.err = status;
     hreq->profile = printDistHybridProfile;
@@ -598,7 +642,15 @@ static int HybridRequest_prepareForExecution(HybridRequest *hreq, RedisModuleCtx
 
     // Construct the command string
     MRCommand xcmd;
-    HybridRequest_buildMRCommand(argv, argc, profileOptions, &xcmd, serialized, sp, &hybridParams);
+    // Get the VectorQuery from the vector request's AST for SHARD_K_RATIO
+    // optimization
+    // Note: parsedVectorData is only set on shards, not on coordinator
+    // The coordinator has the VectorQuery in the AST after parsing
+    const AREQ *vectorRequest = hreq->requests[VECTOR_INDEX];
+    const VectorQuery *vq = (vectorRequest->ast.root && vectorRequest->ast.root->type == QN_VECTOR)
+                      ? vectorRequest->ast.root->vn.vq : NULL;
+    HybridRequest_buildMRCommand(argv, argc, profileOptions, &xcmd, serialized,
+                                 sp, vq, numShards);
 
     xcmd.protocol = HYBRID_RESP_PROTOCOL_VERSION;
     xcmd.forCursor = hreq->reqflags & QEXEC_F_IS_CURSOR;
@@ -650,7 +702,7 @@ static int HybridRequest_executePlan(HybridRequest *hreq, struct ConcurrentCmdCt
 
     // Get the command from the RPNet (it was set during prepareForExecution)
     MRCommand *cmd = &searchRPNet->cmd;
-    int numShards = GetNumShards_UnSafe();
+    int numShards = ConcurrentCmdCtx_GetNumShards(cmdCtx);
     cmd->coordStartTime = hreq->profileClocks.coordStartTime;
 
     const RSOomPolicy oomPolicy = hreq->reqConfig.oomPolicy;
@@ -748,7 +800,10 @@ void RSExecDistHybrid(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
     // Store coordinator start time for dispatch time tracking
     hreq->profileClocks.coordStartTime = ConcurrentCmdCtx_GetCoordStartTime(cmdCtx);
 
-    if (HybridRequest_prepareForExecution(hreq, ctx, argv, argc, sp, &status) != REDISMODULE_OK) {
+    // Get numShards captured from main thread for thread-safe access
+    size_t numShards = ConcurrentCmdCtx_GetNumShards(cmdCtx);
+
+    if (HybridRequest_prepareForExecution(hreq, ctx, argv, argc, sp, numShards, &status) != REDISMODULE_OK) {
       DistHybridCleanups(ctx, cmdCtx, sp, &strong_ref, hreq, reply, &status);
       return;
     }

diff --git a/src/coord/hybrid/dist_hybrid.h b/src/coord/hybrid/dist_hybrid.h
@@ -17,15 +17,19 @@ extern "C" {
 #include "rmr/command.h"
 #include "dist_plan.h"
 #include "profile/options.h"
+#include "vector_index.h"
 
 void RSExecDistHybrid(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
                         struct ConcurrentCmdCtx *cmdCtx);
 
 // For testing purposes
+// numShards is passed from the main thread to ensure thread-safe access
 void HybridRequest_buildMRCommand(RedisModuleString **argv, int argc,
                             ProfileOptions profileOptions,
                             MRCommand *xcmd, arrayof(char*) serialized,
-                            IndexSpec *sp, HybridPipelineParams *hybridParams);
+                            IndexSpec *sp,
+                            const VectorQuery *vq,
+                            size_t numShards);
 
 #ifdef __cplusplus
 }

diff --git a/src/hybrid/parse_hybrid.c b/src/hybrid/parse_hybrid.c
@@ -21,6 +21,8 @@
 #include "spec.h"
 #include "param.h"
 #include "rmalloc.h"
+#include "config.h"
+#include "shard_window_ratio.h"
 
 #include "rmutil/args.h"
 #include "rmutil/rm_assert.h"
@@ -121,6 +123,34 @@ static int parseSearchSubquery(ArgsCursor *ac, AREQ *sreq, QueryError *status) {
   return REDISMODULE_OK;
 }
 
+static int parseShardKRatioClause(ArgsCursor *ac, ParsedVectorData *pvd,
+                                  QueryError *status) {
+  // VSIM @vectorfield vector KNN <nargs> ... SHARD_K_RATIO <ratio>
+  //                                          ^
+  if (AC_IsAtEnd(ac)) {
+    QueryError_SetError(status, QUERY_ERROR_CODE_PARSE_ARGS, "Missing argument value for SHARD_K_RATIO");
+    return REDISMODULE_ERR;
+  }
+
+  const char *ratioStr;
+  size_t ratioStrLen;
+  ratioStr = AC_GetStringNC(ac, &ratioStrLen);
+  double shardKRatio;
+  if (!validateShardKRatio(ratioStr, &shardKRatio, status)) {
+    return REDISMODULE_ERR;
+  }
+
+  // Add as QueryAttribute (will be applied to VectorQuery via QueryNode_ApplyAttributes)
+  QueryAttribute attr = {
+    .name = SHARD_K_RATIO_ATTR,
+    .namelen = strlen(SHARD_K_RATIO_ATTR),
+    .value = rm_strndup(ratioStr, ratioStrLen),
+    .vallen = ratioStrLen
+  };
+  pvd->attributes = array_ensure_append_1(pvd->attributes, attr);
+  return REDISMODULE_OK;
+}
+
 static int parseKNNClause(ArgsCursor *ac, VectorQuery *vq, ParsedVectorData *pvd, QueryError *status) {
   // VSIM @vectorfield vector KNN ...
   //                              ^
@@ -139,6 +169,7 @@ static int parseKNNClause(ArgsCursor *ac, VectorQuery *vq, ParsedVectorData *pvd
   }
 
   bool hasEF = false;
+  bool hasShardKRatio = false;
   RS_ASSERT(pvd->vectorScoreFieldAlias == NULL);
 
   for (int i=0; i<argumentCount; i+=2) {
@@ -179,6 +210,16 @@ static int parseKNNClause(ArgsCursor *ac, VectorQuery *vq, ParsedVectorData *pvd
       addVectorQueryParam(vq, VECSIM_EFRUNTIME, strlen(VECSIM_EFRUNTIME), value, valueLen);
       hasEF = true;
 
+    } else if (AC_AdvanceIfMatch(ac, "SHARD_K_RATIO")) {
+      if (hasShardKRatio) {
+        QueryError_SetError(status, QUERY_ERROR_CODE_DUP_PARAM, "Duplicate SHARD_K_RATIO argument");
+        return REDISMODULE_ERR;
+      }
+      if (parseShardKRatioClause(ac, pvd, status) != REDISMODULE_OK) {
+        return REDISMODULE_ERR;
+      }
+      hasShardKRatio = true;
+
     } else {
       const char *current;
       AC_GetString(ac, &current, NULL, AC_F_NOADVANCE);
@@ -452,8 +493,8 @@ static int parseVectorSubquery(ArgsCursor *ac, AREQ *vreq, QueryError *status) {
     }
   }
 
-  // Check for optional YIELD_SCORE_AS clause
-  if (AC_AdvanceIfMatch(ac, "YIELD_SCORE_AS")) {
+  // Check for optional YIELD_SCORE_AS detecting duplicate arguments
+  while (!AC_IsAtEnd(ac) && AC_AdvanceIfMatch(ac, "YIELD_SCORE_AS")) {
     if (parseYieldScoreClause(ac, pvd, status) != REDISMODULE_OK) {
       goto error;
     }
@@ -484,6 +525,7 @@ static int parseVectorSubquery(ArgsCursor *ac, AREQ *vreq, QueryError *status) {
     case VECSIM_QT_KNN:
       vq->knn.vector = (void*)vectorParam;
       vq->knn.vecLen = vectorParamLen;
+      vq->knn.shardWindowRatio = DEFAULT_SHARD_WINDOW_RATIO;
       break;
     case VECSIM_QT_RANGE:
       vq->range.vector = (void*)vectorParam;

diff --git a/src/hybrid/vector_query_utils.h b/src/hybrid/vector_query_utils.h
@@ -25,7 +25,7 @@ extern "C" {
 typedef struct {
   VectorQuery *query;
   const char *fieldName;       // Field name for later resolution (NOT owned - points to args)
-  QueryAttribute *attributes;  // Non-vector-specific attributes like YIELD_SCORE_AS (OWNED)
+  QueryAttribute *attributes;  // Non-vector-specific attributes like YIELD_SCORE_AS, SHARD_K_RATIO (OWNED)
   bool isParameter;            // true if vector data is a parameter
   bool hasExplicitK;           // Flag to track if K was explicitly set in KNN query
   char *vectorScoreFieldAlias; // Alias for the vector score field (OWNED) - NULL if not explicitly set

diff --git a/src/module.c b/src/module.c
@@ -3668,7 +3668,8 @@ int DistHybridCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
 
   ConcurrentSearchHandlerCtx handlerCtx = {
     .coordStartTime = coordInitialTime,
-    .spec_ref = StrongRef_Demote(spec_ref)
+    .spec_ref = StrongRef_Demote(spec_ref),
+    .numShards = NumShards  // Capture NumShards from main thread for thread-safe access
   };
 
   return ConcurrentSearch_HandleRedisCommandEx(DIST_THREADPOOL, dist_callback, ctx, argv, argc,

diff --git a/src/query.c b/src/query.c
@@ -48,6 +48,7 @@
 #include "iterators/hybrid_reader.h"
 #include "iterators/optimizer_reader.h"
 #include "search_disk.h"
+#include "shard_window_ratio.h"
 #include "idf.h"
 #include "index_result/query_term/query_term.h"
 
@@ -2134,29 +2135,12 @@ int QueryNode_ForEach(QueryNode *q, QueryNode_ForEachCallback callback, void *ct
   return retVal;
 }
 
-static int ValidateShardKRatio(const char *value, double *ratio, QueryError *status) {
-  if (!ParseDouble(value, ratio, 1)) {
-    QueryError_SetWithUserDataFmt(status, QUERY_ERROR_CODE_INVAL,
-      "Invalid shard k ratio value", " '%s'", value);
-    return 0;
-  }
-
-  if (*ratio <= MIN_SHARD_WINDOW_RATIO || *ratio > MAX_SHARD_WINDOW_RATIO) {
-    QueryError_SetWithoutUserDataFmt(status, QUERY_ERROR_CODE_INVAL,
-      "Invalid shard k ratio value: Shard k ratio must be greater than %g and at most %g (got %g)",
-      MIN_SHARD_WINDOW_RATIO, MAX_SHARD_WINDOW_RATIO, *ratio);
-    return 0;
-  }
-
-  return 1;
-}
-
 // Convert the query attribute into a raw vector param to be resolved by the vector iterator
 // down the road. return 0 in case of an unrecognized parameter.
 static int QueryVectorNode_ApplyAttribute(VectorQuery *vq, QueryAttribute *attr, QueryError *status) {
   if (STR_EQCASE(attr->name, attr->namelen, SHARD_K_RATIO_ATTR)) {
     double ratio;
-    if (!ValidateShardKRatio(attr->value, &ratio, status)) {
+    if (!validateShardKRatio(attr->value, &ratio, status)) {
       return 0;
     }
     vq->knn.shardWindowRatio = ratio;