Expand type coverage; fix corrupt large-column-subset encoding

Adds a vector/duration column scenario (single-cell types, overwritten and
null-overwritten across sstables) plus a support-matrix pin confirming they're inside the
supported surface, and >64-regular-column scenarios that byte-compare the large-subset
wire format between the cursor and iterator paths for the first time.

The >64-column scenario immediately exposed real output corruption: SSTableCursorWriter's
encodeLargeColumnsSubset had two bugs against the reference Columns.Serializer format.
First, its trailing-present loop was bounded by the last missing index instead of the
superset size — vacuously empty — so present columns sorting after the last missing
column were dropped from the encoding, and the deserializer then consumed row-body bytes
as column indices: desync, EOF, an sstable that crashes every subsequent read. Second,
encoding-mode selection used missing > supersetCount/2 where both the reference
serializer and the deserializer use presentCount < supersetCount/2 — equivalent for even
superset sizes but flipped for odd sizes at the boundary, making the decoder read the
wrong mode. No pre-existing test exercised present-index mode (few-missing rows always
take missing-index mode), which is how both survived.

over64Columns pins the dropped-tail shape (fails pre-fix with CorruptSSTableException);
over64ColumnsOddSupersetBoundary pins the mode boundary with a 71-column superset and
rows at 34/35/36 present columns.

src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java

@@ -529,24 +529,30 @@ public class SSTableCursorWriter implements AutoCloseable
 
     private void encodeLargeColumnsSubset() throws IOException
     {
-        // no columns are present, nothing to write
         rowHeaderBuffer.writeUnsignedVInt32(missingColumns.size());
-        if (missingColumns.size() > columns.length / 2)
+        // Mode selection must mirror Columns.Serializer.serializeLargeSubset AND its
+        // deserializer exactly: present-index mode iff presentCount < supersetCount / 2.
+        // The previous condition (missing > supersetCount / 2) agreed for even superset
+        // sizes but flipped the mode for odd sizes at missing == supersetCount/2 + 1,
+        // which the deserializer then read in the WRONG mode — corrupted output.
+        int presentCount = columns.length - missingColumns.size();
+        if (presentCount < columns.length / 2)
         {
-            // write present columns
+            // write present column indices: the gaps between missing indices, INCLUDING the
+            // tail after the last missing index — the previous tail loop's bound was the
+            // last missing index itself (vacuously empty), so present columns sorting after
+            // the last missing one were silently dropped from the encoding and the
+            // deserializer consumed row-body bytes as column indices — corrupted output
             int presentIndex = 0;
-            int missingIndex = 0;
             for (int i = 0; i < missingColumns.size(); i++)
             {
-                missingIndex = missingColumns.get(i);
+                int missingIndex = missingColumns.get(i);
                 for (; presentIndex < missingIndex; presentIndex++)
                     rowHeaderBuffer.writeUnsignedVInt32(presentIndex);
                 presentIndex = missingIndex + 1;
             }
-            if (missingIndex < columns.length-1) {
-                for (; presentIndex < missingIndex; presentIndex++)
-                    rowHeaderBuffer.writeUnsignedVInt32(presentIndex);
-            }
+            for (; presentIndex < columns.length; presentIndex++)
+                rowHeaderBuffer.writeUnsignedVInt32(presentIndex);
         }
         else
         {

test/unit/org/apache/cassandra/db/compaction/differential/CursorSupportMatrixTest.java

@@ -101,6 +101,14 @@ public class CursorSupportMatrixTest extends CQLTester
         assertUnsupported("CREATE TABLE %s (pk bigint, ck bigint, u " + udt + ", PRIMARY KEY (pk, ck))");
     }
 
+    /** Vector and duration are inside the supported surface (single-cell types). */
+    @Test
+    public void vectorAndDurationSupported()
+    {
+        assertSupported("CREATE TABLE %s (pk bigint, ck bigint, vec vector<float, 3>, dur duration, " +
+                        "PRIMARY KEY (pk, ck))");
+    }
+
     /** Increment 5 flips this to supported. */
     @Test
     public void countersUnsupported()

test/unit/org/apache/cassandra/db/compaction/differential/EdgeCaseDifferentialCompactionTest.java

@@ -539,4 +539,114 @@ public class EdgeCaseDifferentialCompactionTest extends DifferentialCompactionTe
 
         assertCursorMatchesIteratorAcrossGenerations(cfs, ALLOWLIST);
     }
+
+    /** Vector and duration columns: fixed-dimension float vectors and the
+     *  variable-length duration encoding as ordinary single cells, overwritten and
+     *  null-overwritten (cell tombstone) across sstables. */
+    @Test
+    public void vectorAndDuration() throws Exception
+    {
+        createTable("CREATE TABLE %s (pk bigint, ck bigint, vec vector<float, 3>, dur duration, v text, " +
+                    "PRIMARY KEY (pk, ck))");
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+        cfs.disableAutoCompaction();
+
+        for (long ck = 0; ck < 10; ck++)
+            execute("INSERT INTO %s (pk, ck, vec, dur, v) VALUES (?, ?, [1.5, 2.5, " + ck + ".0], 2h30m, ?)",
+                    1L, ck, "v" + ck);
+        flush();
+
+        for (long ck = 0; ck < 5; ck++)
+            execute("INSERT INTO %s (pk, ck, vec, dur, v) VALUES (?, ?, [9.0, 8.0, 7.0], 45s500ms, ?)",
+                    1L, ck, "w" + ck);
+        // null overwrites: cell tombstones for vector and duration cells
+        execute("INSERT INTO %s (pk, ck, vec, dur) VALUES (?, ?, null, null)", 1L, 7L);
+        flush();
+
+        assertCursorMatchesIteratorAcrossGenerations(cfs, ALLOWLIST);
+    }
+
+    /**
+     * More than 64 regular columns: rows lacking columns switch from the 64-bit-mask
+     * column-subset encoding to the structurally different large-subset wire format —
+     * byte-compared here for the first time; previously only the old simple-suite
+     * exercised it without comparing the paths against each other.
+     */
+    @Test
+    public void over64Columns() throws Exception
+    {
+        StringBuilder ddl = new StringBuilder("CREATE TABLE %s (pk bigint, ck bigint");
+        for (int i = 0; i < 70; i++)
+            ddl.append(", c").append(i).append(" int");
+        ddl.append(", PRIMARY KEY (pk, ck))");
+        createTable(ddl.toString());
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+        cfs.disableAutoCompaction();
+
+        // sparse rows: each ck sets a sliding 10-column window (subset encoding for >64 columns)
+        for (int round = 0; round < 2; round++)
+        {
+            for (long ck = 0; ck < 14; ck++)
+            {
+                StringBuilder stmt = new StringBuilder("INSERT INTO %s (pk, ck");
+                int base = (int) ck * 5 + round * 3;
+                for (int i = 0; i < 10; i++)
+                    stmt.append(", c").append((base + i) % 70);
+                stmt.append(") VALUES (?, ?");
+                for (int i = 0; i < 10; i++)
+                    stmt.append(", ").append(base + i);
+                stmt.append(')');
+                execute(stmt.toString(), 1L, ck);
+            }
+            // one full row per round: the HAS_ALL_COLUMNS path next to large subsets
+            StringBuilder full = new StringBuilder("INSERT INTO %s (pk, ck");
+            for (int i = 0; i < 70; i++)
+                full.append(", c").append(i);
+            full.append(") VALUES (?, ?");
+            for (int i = 0; i < 70; i++)
+                full.append(", ").append(i);
+            full.append(')');
+            execute(full.toString(), 1L, 99L);
+            flush();
+        }
+
+        assertCursorMatchesIteratorAcrossGenerations(cfs, ALLOWLIST);
+    }
+
+    /**
+     * ODD superset size at the subset-encoding mode boundary: with 71 columns, the encoder
+     * and decoder must agree on present-index vs missing-index
+     * mode at exactly presentCount == 35 (the integer-division boundary of supersetCount/2).
+     * Rows at 34/35/36 present columns straddle the boundary from both sides.
+     */
+    @Test
+    public void over64ColumnsOddSupersetBoundary() throws Exception
+    {
+        StringBuilder ddl = new StringBuilder("CREATE TABLE %s (pk bigint, ck bigint");
+        for (int i = 0; i < 71; i++)
+            ddl.append(", c").append(i).append(" int");
+        ddl.append(", PRIMARY KEY (pk, ck))");
+        createTable(ddl.toString());
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+        cfs.disableAutoCompaction();
+
+        for (int round = 0; round < 2; round++)
+        {
+            long ck = 0;
+            for (int present : new int[]{ 34, 35, 36, 70 })
+            {
+                StringBuilder stmt = new StringBuilder("INSERT INTO %s (pk, ck");
+                for (int i = 0; i < present; i++)
+                    stmt.append(", c").append((i + round) % 71); // shift per round so the merge unions
+                stmt.append(") VALUES (?, ?");
+                for (int i = 0; i < present; i++)
+                    stmt.append(", ").append(i);
+                stmt.append(')');
+                execute(stmt.toString(), 1L, ck++);
+            }
+            flush();
+        }
+
+        assertCursorMatchesIteratorAcrossGenerations(cfs, ALLOWLIST);
+    }
 }