[GG] fix(dcp): build one-shard indexer query-split topology

tests/distributed/test_indexer_parallel_groups.py

@@ -8,6 +8,7 @@
 from vllm.distributed import parallel_state
 from vllm.distributed.parallel_state import (
     _build_indexer_replica_group_ranks,
+    _needs_indexer_replica_groups,
     _validate_indexer_shard_count,
 )
 
@@ -54,6 +55,14 @@ def test_build_indexer_groups_cover_dcp1_partial_and_full(
     assert query_split_groups == expected_query_split
 
 
+@pytest.mark.parametrize(
+    ("indexer_shards", "expected"),
+    [(0, False), (1, True), (2, True), (4, True), (8, False)],
+)
+def test_indexer_replica_group_gate_for_dcp8(indexer_shards, expected):
+    assert _needs_indexer_replica_groups(indexer_shards, 8) is expected
+
+
 def test_build_indexer_replica_groups_stay_inside_each_tp_group():
     dcp_groups, query_split_groups = _build_indexer_replica_group_ranks(
         [list(range(8)), list(range(8, 16))], 4

vllm/distributed/parallel_state.py

@@ -1542,6 +1542,11 @@ def _validate_indexer_shard_count(indexer_shards: int, dcp_size: int) -> None:
         )
 
 
+def _needs_indexer_replica_groups(indexer_shards: int, dcp_size: int) -> bool:
+    """Return whether the indexer needs replica-specific process groups."""
+    return 1 <= indexer_shards < dcp_size
+
+
 _DCP_CKV_PREFETCH: GroupCoordinator | None = None
 
 
@@ -2101,7 +2106,9 @@ def initialize_model_parallel(
     )
     indexer_shards = int(envs.VLLM_DCP_INDEXER_SHARDS)
     _validate_indexer_shard_count(indexer_shards, decode_context_model_parallel_size)
-    if 1 < indexer_shards < decode_context_model_parallel_size:
+    if _needs_indexer_replica_groups(
+        indexer_shards, decode_context_model_parallel_size
+    ):
         indexer_dcp_ranks, indexer_query_split_ranks = (
             _build_indexer_replica_group_ranks(tp_group_ranks, indexer_shards)
         )