diff --git a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java
index 0c5b9eacdf45..ae719a96fb23 100644
--- a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java
+++ b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java
@@ -85,6 +85,7 @@ public ColumnCapabilities getColumnCapabilities(String column)
   // Bumped on every setDelegate(...) so per-call selector wrappers can detect group transitions and rebuild their
   // cached inner state
   private long generation;
+  private final RowIdSupplier rowIdSupplier = new DelegatingRowIdSupplier(this);
 
   public ClusteringColumnSelectorFactory(
       ColumnSelectorFactory delegate,
@@ -177,7 +178,8 @@ public ColumnCapabilities getColumnCapabilities(String column)
   @Override
   public RowIdSupplier getRowIdSupplier()
   {
-    return delegate.getRowIdSupplier();
+    // A delegate may not support row-id caching; mirror that so callers skip caching (null)
+    return delegate.getRowIdSupplier() == null ? null : rowIdSupplier;
   }
 
   Object currentValue(int idx)
@@ -185,6 +187,44 @@ Object currentValue(int idx)
     return clusteringValues[idx];
   }
 
+  /**
+   * A {@link RowIdSupplier} bound to the factory, {@link #getRowId} follows whichever delegate is current, so a
+   * supplier grabbed once keeps tracking the active group across {@code ConcatenatingCursor} transitions.
+   */
+  private static final class DelegatingRowIdSupplier implements RowIdSupplier
+  {
+    private final ClusteringColumnSelectorFactory parent;
+    private long rowId = INIT;
+    private long lastDelegateRowId = INIT;
+    private RowIdSupplier lastDelegate;
+
+    private DelegatingRowIdSupplier(ClusteringColumnSelectorFactory parent)
+    {
+      this.parent = parent;
+    }
+
+    @Override
+    public long getRowId()
+    {
+      final RowIdSupplier delegate = parent.getDelegate().getRowIdSupplier();
+      if (delegate == null) {
+        // getRowIdSupplier() only hands out this wrapper when the delegate supports row ids; clustered groups are
+        // homogeneous, so a null here would mean a group mid-cursor stopped supporting them.
+        throw DruidException.defensive("delegate row id supplier became null across a cluster-group transition");
+      }
+      final long id = delegate.getRowId();
+      if (id == INIT) {
+        return INIT;
+      }
+      if (delegate != lastDelegate || id != lastDelegateRowId) {
+        lastDelegate = delegate;
+        lastDelegateRowId = id;
+        rowId++;
+      }
+      return rowId;
+    }
+  }
+
   /**
    * Dimension selector for a clustering column. Delegates the value lookup back to the parent factory each call so
    * that group transitions (which mutate the parent's clustering values) are observed immediately. Internally
diff --git a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java
index 722e1017e9ac..e82c0a767221 100644
--- a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java
+++ b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java
@@ -58,6 +58,7 @@ public class ClusteringVectorColumnSelectorFactory implements VectorColumnSelect
   // Bumped on every setDelegate(...) so per-call selector wrappers can detect group transitions and rebuild their
   // cached inner state.
   private long generation;
+  private final ReadableVectorInspector readableVectorInspector = new DelegatingReadableVectorInspector(this);
 
   /**
    * Convenience overload that derives {@code maxVectorSize} from the supplied delegate. Used by single-group
@@ -120,7 +121,7 @@ Object currentValue(int idx)
   @Override
   public ReadableVectorInspector getReadableVectorInspector()
   {
-    return delegate.getReadableVectorInspector();
+    return readableVectorInspector;
   }
 
   @Override
@@ -201,6 +202,52 @@ public ColumnCapabilities getColumnCapabilities(String column)
     return ColumnCapabilitiesImpl.createSimpleNumericColumnCapabilities(type);
   }
 
+  /**
+   * A {@link ReadableVectorInspector} bound to the factory, not to one delegate: {@link #getCurrentVectorSize} and
+   * {@link #getId} follow whichever delegate is current, so an inspector grabbed once keeps tracking the active group
+   * across {@code ConcatenatingVectorCursor} transitions. {@link #getMaxVectorSize} is the factory's fixed max.
+   */
+  private static final class DelegatingReadableVectorInspector implements ReadableVectorInspector
+  {
+    private final ClusteringVectorColumnSelectorFactory parent;
+    private int vectorId = NULL_ID;
+    private int lastDelegateId = NULL_ID;
+    private VectorColumnSelectorFactory lastDelegate;
+
+    private DelegatingReadableVectorInspector(ClusteringVectorColumnSelectorFactory parent)
+    {
+      this.parent = parent;
+    }
+
+    @Override
+    public int getMaxVectorSize()
+    {
+      return parent.getMaxVectorSize();
+    }
+
+    @Override
+    public int getCurrentVectorSize()
+    {
+      return parent.getDelegate().getReadableVectorInspector().getCurrentVectorSize();
+    }
+
+    @Override
+    public int getId()
+    {
+      final VectorColumnSelectorFactory delegate = parent.getDelegate();
+      final int id = delegate.getReadableVectorInspector().getId();
+      if (id == NULL_ID) {
+        return NULL_ID;
+      }
+      if (delegate != lastDelegate || id != lastDelegateId) {
+        lastDelegate = delegate;
+        lastDelegateId = id;
+        vectorId++;
+      }
+      return vectorId;
+    }
+  }
+
   private static final class ClusteringVectorValueSelector implements VectorValueSelector
   {
     private final ClusteringVectorColumnSelectorFactory parent;
diff --git a/processing/src/test/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactoryTest.java b/processing/src/test/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactoryTest.java
index e5181c970bba..1987c8768573 100644
--- a/processing/src/test/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactoryTest.java
+++ b/processing/src/test/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactoryTest.java
@@ -200,6 +200,136 @@ void testNonClusteringSelectorObservesNewDelegateAfterSetDelegate()
     Assertions.assertEquals("region", second.lastDimSelectorName);
   }
 
+  @Test
+  void testRowIdSupplierFollowsCurrentDelegate()
+  {
+    // getRowIdSupplier() must return a supplier that reflects the current delegate, even for a reference grabbed
+    // before a group transition. A multi-group ConcatenatingCursor swaps the delegate on each group, and a consumer
+    // that cached the supplier must observe the active group's row id (not the group current when it first asked).
+    RecordingDelegate first = new RecordingDelegate();
+    first.rowId = 7;
+    ClusteringColumnSelectorFactory f = new ClusteringColumnSelectorFactory(first, SIGNATURE, new Object[]{"acme"});
+
+    RowIdSupplier supplier = f.getRowIdSupplier();
+    Assertions.assertNotNull(supplier);
+    long firstId = supplier.getRowId();
+
+    // Simulate a group transition; the second group's offset restarts its row id at 0.
+    RecordingDelegate second = new RecordingDelegate();
+    second.rowId = 0;
+    f.setDelegate(second, new Object[]{"globex"});
+
+    // The previously-acquired supplier now reads the second delegate, and its remapped id differs despite the reset.
+    Assertions.assertNotEquals(firstId, supplier.getRowId());
+  }
+
+  @Test
+  void testGetRowIdUniqueAcrossSingleRowGroups()
+  {
+    // A group whose offset restarts at 0, followed by another such group, must not report the same row id across the
+    // transition, or a single-slot row-id cache would hand back the previous group's stale row.
+    ClusteringColumnSelectorFactory f = new ClusteringColumnSelectorFactory(
+        new RecordingDelegate(),   // rowId 0
+        SIGNATURE,
+        new Object[]{"a"}
+    );
+    RowIdSupplier supplier = f.getRowIdSupplier();
+    long idA = supplier.getRowId();
+
+    f.setDelegate(new RecordingDelegate(), new Object[]{"b"});   // rowId 0
+    long idB = supplier.getRowId();
+
+    f.setDelegate(new RecordingDelegate(), new Object[]{"c"});   // rowId 0
+    long idC = supplier.getRowId();
+
+    Assertions.assertNotEquals(idA, idB);
+    Assertions.assertNotEquals(idB, idC);
+    Assertions.assertNotEquals(idA, idC);
+  }
+
+  @Test
+  void testGetRowIdAdvancesWithinGroupAndStaysAheadAcrossTransition()
+  {
+    // Within a group, row ids track the delegate offset as it advances. Across a transition, the next group's first
+    // row id must be strictly greater than the last id the previous group handed out.
+    RecordingDelegate d = new RecordingDelegate();
+    ClusteringColumnSelectorFactory f = new ClusteringColumnSelectorFactory(d, SIGNATURE, new Object[]{"a"});
+    RowIdSupplier supplier = f.getRowIdSupplier();
+    d.rowId = 0;
+    long id0 = supplier.getRowId();
+    d.rowId = 5;
+    long id1 = supplier.getRowId();
+    d.rowId = 9;
+    long id2 = supplier.getRowId();
+    Assertions.assertTrue(id0 < id1 && id1 < id2, "row ids must increase as the group advances");
+
+    f.setDelegate(new RecordingDelegate(), new Object[]{"b"});   // rowId 0
+    Assertions.assertTrue(supplier.getRowId() > id2, "new group's first row id must exceed the previous group's last");
+  }
+
+  @Test
+  void testGetRowIdMintsMonotonicIdsRegardlessOfDelegateIdOrder()
+  {
+    // The RowIdSupplier contract does not require monotonic (or meaningful) delegate ids. We mint our own
+    // strictly-increasing ids purely from change detection, so a non-monotonic sequence within a group and an id that
+    // overlaps across a group boundary both still yield clean, collision-free ids.
+    RecordingDelegate d = new RecordingDelegate();
+    ClusteringColumnSelectorFactory f = new ClusteringColumnSelectorFactory(d, SIGNATURE, new Object[]{"a"});
+    RowIdSupplier supplier = f.getRowIdSupplier();
+
+    d.rowId = 100;
+    long a = supplier.getRowId();
+    d.rowId = 5;                       // non-monotonic within the group
+    long b = supplier.getRowId();
+
+    RecordingDelegate second = new RecordingDelegate();
+    second.rowId = 100;                // reuse an id the first group already handed out
+    f.setDelegate(second, new Object[]{"b"});
+    long c = supplier.getRowId();
+
+    Assertions.assertTrue(a < b && b < c, "minted row ids must be strictly increasing regardless of delegate id order");
+  }
+
+  @Test
+  void testGetRowIdSupplierNullWhenDelegateHasNone()
+  {
+    // A delegate that doesn't support row-id caching (null supplier) must surface as a null supplier on the wrapper,
+    // so callers take the no-caching path rather than caching against a fabricated id.
+    RecordingDelegate d = new RecordingDelegate();
+    d.supportsRowId = false;
+    ClusteringColumnSelectorFactory f = new ClusteringColumnSelectorFactory(d, SIGNATURE, new Object[]{"a"});
+    Assertions.assertNull(f.getRowIdSupplier());
+  }
+
+  @Test
+  void testGetRowIdStableWhenPositionUnchanged()
+  {
+    // The caching contract: reading twice without the cursor moving (same delegate, same raw id) must return the SAME
+    // minted id, so a downstream row-id-keyed cache hits instead of recomputing. Guards against an always-tick regression.
+    RecordingDelegate d = new RecordingDelegate();
+    d.rowId = 3;
+    ClusteringColumnSelectorFactory f = new ClusteringColumnSelectorFactory(d, SIGNATURE, new Object[]{"a"});
+    RowIdSupplier supplier = f.getRowIdSupplier();
+    long first = supplier.getRowId();
+    Assertions.assertEquals(first, supplier.getRowId());
+    Assertions.assertEquals(first, supplier.getRowId());
+  }
+
+  @Test
+  void testGetRowIdReturnsSentinelWhenDelegateUnpositioned()
+  {
+    // When the delegate reports the unpositioned sentinel (INIT), the wrapper must pass it through unchanged and not
+    // advance its minted counter, so a later real row still mints a fresh id starting at 0.
+    RecordingDelegate d = new RecordingDelegate();
+    d.rowId = RowIdSupplier.INIT;
+    ClusteringColumnSelectorFactory f = new ClusteringColumnSelectorFactory(d, SIGNATURE, new Object[]{"a"});
+    RowIdSupplier supplier = f.getRowIdSupplier();
+    Assertions.assertEquals(RowIdSupplier.INIT, supplier.getRowId());
+
+    d.rowId = 0;
+    Assertions.assertEquals(0L, supplier.getRowId());
+  }
+
   @Test
   void testGetColumnCapabilitiesForClusteringColumns()
   {
@@ -389,6 +519,14 @@ private static class RecordingDelegate implements ColumnSelectorFactory
     String lastDimSelectorName;
     String lastValueSelectorName;
     String lastCapabilitiesColumn;
+    // Row id handed out by this delegate's supplier; per-group offsets restart at 0, so the default mirrors a fresh
+    // group cursor. supportsRowId=false models a delegate with no row-id caching (getRowIdSupplier() == null).
+    long rowId;
+    boolean supportsRowId = true;
+    // A single stable supplier instance (reads the mutable rowId each call), mirroring a real factory that returns
+    // `this` from getRowIdSupplier(). A fresh lambda per call would make the wrapper's delegate-identity check tick on
+    // every read, hiding whether the raw-id-change branch works.
+    private final RowIdSupplier stableRowIdSupplier = () -> rowId;
 
     @Override
     public DimensionSelector makeDimensionSelector(org.apache.druid.query.dimension.DimensionSpec dimensionSpec)
@@ -416,7 +554,7 @@ public ColumnCapabilities getColumnCapabilities(String column)
     @Override
     public RowIdSupplier getRowIdSupplier()
     {
-      return null;
+      return supportsRowId ? stableRowIdSupplier : null;
     }
   }
 }
diff --git a/processing/src/test/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactoryTest.java b/processing/src/test/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactoryTest.java
index 6e1dfbf8bcb1..4b9106dc6722 100644
--- a/processing/src/test/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactoryTest.java
+++ b/processing/src/test/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactoryTest.java
@@ -21,6 +21,7 @@
 
 import org.apache.druid.error.DruidException;
 import org.apache.druid.query.dimension.DefaultDimensionSpec;
+import org.apache.druid.query.dimension.DimensionSpec;
 import org.apache.druid.segment.column.ColumnCapabilities;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
@@ -291,11 +292,183 @@ void testMultiValueDimensionSelectorRejected()
     );
   }
 
+  @Test
+  void testReadableVectorInspectorFollowsCurrentDelegate()
+  {
+    // getReadableVectorInspector() must return an inspector that reflects the current delegate, even for a reference
+    // grabbed before a group transition. A multi-group ConcatenatingVectorCursor swaps the delegate on each group, and
+    // a consumer that cached the inspector must observe the active group's size (not the group that was current when
+    // it first asked).
+    ClusteringVectorColumnSelectorFactory f = new ClusteringVectorColumnSelectorFactory(
+        new StubDelegate(new NoFilterVectorOffset(8, 0, 5)),   // current vector size 5
+        CLUSTER_SIGNATURE,
+        new Object[]{"acme"}
+    );
+
+    ReadableVectorInspector inspector = f.getReadableVectorInspector();
+    Assertions.assertEquals(8, inspector.getMaxVectorSize());
+    Assertions.assertEquals(5, inspector.getCurrentVectorSize());
+    int firstId = inspector.getId();
+
+    // Simulate a group transition. The second group's offset restarts its id at 0 and has a smaller final vector.
+    f.setDelegate(new StubDelegate(new NoFilterVectorOffset(8, 0, 3)), new Object[]{"globex"});
+
+    // The previously-acquired inspector reference now reports the second group's size ...
+    Assertions.assertEquals(3, inspector.getCurrentVectorSize());
+    // ... and a distinct id, even though the delegate offset restarted at 0.
+    Assertions.assertNotEquals(firstId, inspector.getId());
+  }
+
+  @Test
+  void testGetIdUniqueAcrossSingleVectorGroups()
+  {
+    // A group that fits in one vector reports delegate id 0; the next group's offset also restarts at 0. The remapped
+    // id must differ across the transition so a single-slot id cache doesn't treat the new group's first vector as
+    // unchanged and hand back the previous group's stale vector.
+    ClusteringVectorColumnSelectorFactory f = new ClusteringVectorColumnSelectorFactory(
+        new StubDelegate(new NoFilterVectorOffset(8, 0, 4)),
+        CLUSTER_SIGNATURE,
+        new Object[]{"a"}
+    );
+    ReadableVectorInspector inspector = f.getReadableVectorInspector();
+    int idA = inspector.getId();
+
+    f.setDelegate(new StubDelegate(new NoFilterVectorOffset(8, 0, 4)), new Object[]{"b"});
+    int idB = inspector.getId();
+
+    f.setDelegate(new StubDelegate(new NoFilterVectorOffset(8, 0, 4)), new Object[]{"c"});
+    int idC = inspector.getId();
+
+    Assertions.assertNotEquals(idA, idB);
+    Assertions.assertNotEquals(idB, idC);
+    Assertions.assertNotEquals(idA, idC);
+  }
+
+  @Test
+  void testGetIdAdvancesWithinGroupAndStaysAheadAcrossTransition()
+  {
+    // Within a group, ids track the delegate offset as it advances. Across a transition, the next group's first id
+    // must be strictly greater than the last id the previous group handed out.
+    NoFilterVectorOffset offset = new NoFilterVectorOffset(4, 0, 12); // ids 0, 4, 8 as it advances
+    ClusteringVectorColumnSelectorFactory f = new ClusteringVectorColumnSelectorFactory(
+        new StubDelegate(offset),
+        CLUSTER_SIGNATURE,
+        new Object[]{"a"}
+    );
+    ReadableVectorInspector inspector = f.getReadableVectorInspector();
+    int id0 = inspector.getId();
+    offset.advance();
+    int id1 = inspector.getId();
+    offset.advance();
+    int id2 = inspector.getId();
+    Assertions.assertTrue(id0 < id1 && id1 < id2, "ids must increase as the group advances");
+
+    f.setDelegate(new StubDelegate(new NoFilterVectorOffset(4, 0, 4)), new Object[]{"b"});
+    Assertions.assertTrue(inspector.getId() > id2, "new group's first id must exceed the previous group's last id");
+  }
+
+  @Test
+  void testGetIdMintsMonotonicIdsRegardlessOfDelegateIdOrder()
+  {
+    // The ReadableVectorInspector contract does not require monotonic (or meaningful) delegate ids. We mint our own
+    // strictly-increasing ids purely from change detection, so a non-monotonic sequence within a group and an id that
+    // overlaps across a group boundary both still yield clean, collision-free ids.
+    SettableVectorInspector first = new SettableVectorInspector(4);
+    ClusteringVectorColumnSelectorFactory f = new ClusteringVectorColumnSelectorFactory(
+        new StubDelegate(first),
+        CLUSTER_SIGNATURE,
+        new Object[]{"a"}
+    );
+    ReadableVectorInspector inspector = f.getReadableVectorInspector();
+
+    first.id = 100;
+    int a = inspector.getId();
+    first.id = 5;                    // non-monotonic within the group
+    int b = inspector.getId();
+
+    SettableVectorInspector second = new SettableVectorInspector(4);
+    second.id = 100;                 // reuse an id the first group already handed out
+    f.setDelegate(new StubDelegate(second), new Object[]{"b"});
+    int c = inspector.getId();
+
+    Assertions.assertTrue(a < b && b < c, "minted ids must be strictly increasing regardless of delegate id order");
+  }
+
+  @Test
+  void testGetIdStableWhenPositionUnchanged()
+  {
+    // The caching contract: reading getId() twice without advancing (same delegate, same raw id) must return the SAME
+    // minted id, so a downstream id-keyed cache hits instead of recomputing. Guards against an always-tick regression.
+    ClusteringVectorColumnSelectorFactory f = new ClusteringVectorColumnSelectorFactory(
+        new StubDelegate(new NoFilterVectorOffset(8, 0, 5)),
+        CLUSTER_SIGNATURE,
+        new Object[]{"a"}
+    );
+    ReadableVectorInspector inspector = f.getReadableVectorInspector();
+    int first = inspector.getId();
+    Assertions.assertEquals(first, inspector.getId());
+    Assertions.assertEquals(first, inspector.getId());
+  }
+
+  @Test
+  void testGetIdReturnsSentinelWhenDelegateUnpositioned()
+  {
+    // When the delegate inspector reports NULL_ID (no stable vector), the wrapper must pass it through unchanged and
+    // not advance its minted counter, so a later real vector still mints a fresh id starting at 0.
+    SettableVectorInspector state = new SettableVectorInspector(4);
+    state.id = ReadableVectorInspector.NULL_ID;
+    ClusteringVectorColumnSelectorFactory f = new ClusteringVectorColumnSelectorFactory(
+        new StubDelegate(state),
+        CLUSTER_SIGNATURE,
+        new Object[]{"a"}
+    );
+    ReadableVectorInspector inspector = f.getReadableVectorInspector();
+    Assertions.assertEquals(ReadableVectorInspector.NULL_ID, inspector.getId());
+
+    state.id = 0;
+    Assertions.assertEquals(0, inspector.getId());
+  }
+
   private static ReadableVectorInspector inspectorFor(int size)
   {
     return new NoFilterVectorOffset(size, 0, size);
   }
 
+  /**
+   * A {@link ReadableVectorInspector} whose id and current size can be set arbitrarily, to exercise delegate id
+   * sequences that are not monotonic (which the contract permits).
+   */
+  private static final class SettableVectorInspector implements ReadableVectorInspector
+  {
+    private final int maxSize;
+    private int id;
+    private int currentSize;
+
+    private SettableVectorInspector(int maxSize)
+    {
+      this.maxSize = maxSize;
+      this.currentSize = maxSize;
+    }
+
+    @Override
+    public int getId()
+    {
+      return id;
+    }
+
+    @Override
+    public int getMaxVectorSize()
+    {
+      return maxSize;
+    }
+
+    @Override
+    public int getCurrentVectorSize()
+    {
+      return currentSize;
+    }
+  }
+
   private static class StubDelegate implements VectorColumnSelectorFactory
   {
     final ReadableVectorInspector inspector;
@@ -317,7 +490,7 @@ public ReadableVectorInspector getReadableVectorInspector()
 
     @Override
     public SingleValueDimensionVectorSelector makeSingleValueDimensionSelector(
-        org.apache.druid.query.dimension.DimensionSpec dimensionSpec
+        DimensionSpec dimensionSpec
     )
     {
       lastSingleValDimRequest = dimensionSpec.getDimension();
@@ -326,7 +499,7 @@ public SingleValueDimensionVectorSelector makeSingleValueDimensionSelector(
 
     @Override
     public MultiValueDimensionVectorSelector makeMultiValueDimensionSelector(
-        org.apache.druid.query.dimension.DimensionSpec dimensionSpec
+        DimensionSpec dimensionSpec
     )
     {
       throw new UnsupportedOperationException("not used");