fix: address Copilot review comments on DRIVER-368

- ScassandraCluster: compute hostIdByNodeCount once as an instance field
  in the constructor so all primeMetadata() calls (init and restart)
  use stable, consistent host_id UUIDs across system.local and
  system.peers rows for the same logical node.

- ControlConnection.intersectWithNeeded(): return null instead of an
  empty set when no server columns overlap the needed set, keeping the
  cache in the uninitialised sentinel state and preventing
  buildProjectedQuery() from generating an invalid empty-column SELECT.

- ControlConnectionUnitTest: update empty-intersection tests to assert
  null (matching new contract); replace String.contains() assertions in
  testBuildProjectedQueryAllColumnsPresent() with exact column-list
  parsing via extractSelectedColumns() to avoid false positives when one
  column name is a substring of another (e.g. native_port vs
  native_transport_port).

Author: nikagra

driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java

@@ -932,8 +932,10 @@ public void onFailure(Throwable t) {
 
   /**
    * Returns the intersection of the columns returned by the server (from {@code rs}) with the given
-   * {@code needed} set. The result is used to cache projected column lists so subsequent queries
-   * fetch only what the driver actually reads.
+   * {@code needed} set, or {@code null} if the intersection is empty. The result is used to cache
+   * projected column lists so subsequent queries fetch only what the driver actually reads. A
+   * {@code null} return keeps the cache in the "uninitialized" sentinel state, ensuring the driver
+   * continues issuing {@code SELECT *} rather than generating an invalid empty-column projection.
    */
   @VisibleForTesting
   static Set<String> intersectWithNeeded(ResultSet rs, ImmutableSet<String> needed) {
@@ -943,7 +945,8 @@ static Set<String> intersectWithNeeded(ResultSet rs, ImmutableSet<String> needed
         result.add(def.getName());
       }
     }
-    return result.build();
+    ImmutableSet<String> built = result.build();
+    return built.isEmpty() ? null : built;
   }
 
   /**

driver-core/src/test/java/com/datastax/driver/core/ControlConnectionUnitTest.java

@@ -175,23 +175,26 @@ public void testIntersectWithNeededHandlesSubset() {
   }
 
   @Test(groups = "unit")
-  public void testIntersectWithNeededNoOverlapReturnsEmpty() {
+  public void testIntersectWithNeededNoOverlapReturnsNull() {
+    // When no server columns match the needed set, the result should be null so the cache remains
+    // in the uninitialized sentinel state (avoids generating an empty-column SELECT projection).
     ImmutableSet<String> needed = ImmutableSet.of("cluster_name", "tokens");
     ResultSet rs = mockResultSetWithColumns("some_other_col", "another_col");
 
     Set<String> result = ControlConnection.intersectWithNeeded(rs, needed);
 
-    assertThat(result).isEmpty();
+    assertThat(result).isNull();
   }
 
   @Test(groups = "unit")
-  public void testIntersectWithNeededEmptyResultSet() {
+  public void testIntersectWithNeededEmptyResultSetReturnsNull() {
+    // An empty ResultSet has no column definitions, so the intersection is empty → null.
     ImmutableSet<String> needed = ControlConnection.LOCAL_COLUMNS_OF_INTEREST;
     ResultSet rs = mockResultSetWithColumns();
 
     Set<String> result = ControlConnection.intersectWithNeeded(rs, needed);
 
-    assertThat(result).isEmpty();
+    assertThat(result).isNull();
   }
 
   // ---------------------------------------------------------------------------
@@ -235,17 +238,37 @@ public void testBuildProjectedQuerySingleColumn() {
 
   @Test(groups = "unit")
   public void testBuildProjectedQueryAllColumnsPresent() {
-    // Every column in the needed set appears somewhere in the resulting query
+    // Every column in the needed set must appear as an exact identifier in the projected SELECT
+    // list. Use exact parsing to avoid false positives where one column name is a substring of
+    // another (e.g. "native_port" inside "native_transport_port").
     Set<String> columns = ControlConnection.PEERS_COLUMNS_OF_INTEREST;
     String query = ControlConnection.buildProjectedQuery("system.peers", columns, null);
+    Set<String> selectedColumns = extractSelectedColumns(query);
 
     for (String col : columns) {
-      assertThat(query).as("query should contain column: " + col).contains(col);
+      assertThat(selectedColumns).as("query should project column: " + col).contains(col);
     }
     assertThat(query).contains(" FROM system.peers");
     assertThat(query).doesNotContain("WHERE");
   }
 
+  /**
+   * Parses the column identifiers from the {@code SELECT col1, col2, ... FROM ...} portion of a
+   * projected query string and returns them as a set of trimmed names.
+   */
+  private Set<String> extractSelectedColumns(String query) {
+    int selectStart = query.indexOf("SELECT ");
+    int fromStart = query.indexOf(" FROM ");
+    assertThat(selectStart).as("query should start with SELECT").isEqualTo(0);
+    assertThat(fromStart).as("query should contain FROM").isGreaterThan(selectStart);
+    String columnList = query.substring("SELECT ".length(), fromStart);
+    ImmutableSet.Builder<String> builder = ImmutableSet.builder();
+    for (String col : columnList.split(",")) {
+      builder.add(col.trim());
+    }
+    return builder.build();
+  }
+
   // ---------------------------------------------------------------------------
   // Cache fields: declared as volatile, private, instance-level Set
   // ---------------------------------------------------------------------------

driver-core/src/test/java/com/datastax/driver/core/ScassandraCluster.java

@@ -82,6 +82,13 @@ public class ScassandraCluster {
 
   private final boolean peersV2;
 
+  /**
+   * One stable UUID per node (keyed by 1-based nodeCount), computed once at construction so that
+   * system.local and system.peers rows for the same node always carry the same host_id regardless
+   * of which Scassandra process is being primed or how many times primeMetadata() is called.
+   */
+  private final Map<Integer, java.util.UUID> hostIdByNodeCount;
+
   ScassandraCluster(
       Integer[] nodes,
       String ipPrefix,
@@ -148,6 +155,17 @@ public class ScassandraCluster {
     instances = instanceListBuilder.build();
     dcNodeMap = dcNodeMapBuilder.build();
 
+    // Compute stable host_id UUIDs once so every primeMetadata() call uses the same values.
+    Map<Integer, java.util.UUID> hostIds = new HashMap<>();
+    int tempCount = 1;
+    for (Integer dc : new TreeSet<Integer>(dcNodeMap.keySet())) {
+      for (int n = 0; n < dcNodeMap.get(dc).size(); n++) {
+        hostIds.put(tempCount, UUIDs.random());
+        tempCount++;
+      }
+    }
+    this.hostIdByNodeCount = hostIds;
+
     // Prime correct keyspace table based on C* version.
     String[] versionArray = this.cassandraVersion.split("\\.|-");
     double major = Double.parseDouble(versionArray[0] + "." + versionArray[1]);
@@ -392,17 +410,6 @@ public List<Long> getTokensForDC(int dc) {
   private void primeMetadata(Scassandra node) {
     PrimingClient client = node.primingClient();
 
-    // Pre-generate one UUID per node so that system.local and system.peers rows for the same node
-    // use the same host_id. Keys are nodeCount values (1-based).
-    Map<Integer, java.util.UUID> hostIdByNodeCount = new HashMap<>();
-    int tempCount = 1;
-    for (Integer dc : new TreeSet<Integer>(dcNodeMap.keySet())) {
-      for (int n = 0; n < dcNodeMap.get(dc).size(); n++) {
-        hostIdByNodeCount.put(tempCount, UUIDs.random());
-        tempCount++;
-      }
-    }
-
     int nodeCount = 1;
 
     ImmutableList.Builder<Map<String, ?>> rows = ImmutableList.builder();