Clean up QWP delta dictionary tests

Author: glasstiger

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictCatchUpTest.java

@@ -31,9 +31,6 @@
 import org.junit.Test;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -329,19 +326,6 @@ private static String symbolName(int i) {
         return "symbol" + (1000 + i);
     }
 
-    private static int readVarint(byte[] buf, int[] pos) {
-        int result = 0;
-        int shift = 0;
-        while (pos[0] < buf.length) {
-            int b = buf[pos[0]++] & 0xFF;
-            result |= (b & 0x7F) << shift;
-            if ((b & 0x80) == 0) return result;
-            shift += 7;
-            if (shift > 28) throw new IllegalStateException("varint too long");
-        }
-        throw new IllegalStateException("varint truncated");
-    }
-
     private static void waitFor(BoolCondition cond, long timeoutMillis) {
         long deadline = System.currentTimeMillis() + timeoutMillis;
         while (System.currentTimeMillis() < deadline) {
@@ -412,9 +396,9 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
             }
             int connNumber = dictsByConn.size();
             List<String> dict = dictsByConn.get(connNumber - 1);
-            accumulate(data, dict);
+            QwpWireTestUtils.accumulateDeltaDictionary(data, dict);
             if (connNumber == 2) {
-                if (tableCount(data) == 0) {
+                if (QwpWireTestUtils.tableCount(data) == 0) {
                     sawZeroTableFrameOnConn2 = true;
                     // FLAG_DEFER_COMMIT is bit 0x01 of the flags byte (offset 5).
                     catchUpDeferredOnConn2 = (data[5] & 0x01) != 0;
@@ -427,7 +411,7 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                     // frame arriving ahead of the new one -- that replay carries its
                     // original deltaStart 0 and does not trip the flag.
                     int[] pos = {12};
-                    if (readVarint(data, pos) >= 1) {
+                    if (QwpWireTestUtils.readVarint(data, pos) >= 1) {
                         conn2SawDeltaAboveBaseline = true;
                     }
                 }
@@ -437,7 +421,7 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                 if (newConnection) {
                     ackSequenceStarts.add(ackSequence);
                 }
-                client.sendBinary(buildAck(ackSequence));
+                client.sendBinary(QwpWireTestUtils.buildAck(ackSequence));
                 // Drop the first connection right after ACKing its only frame,
                 // forcing the sender to reconnect onto a fresh dictionary.
                 if (connNumber == 1) {
@@ -451,41 +435,6 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
             }
         }
 
-        private static void accumulate(byte[] frame, List<String> dict) {
-            final byte FLAG_DELTA_SYMBOL_DICT = 0x08;
-            if (frame.length < 12 || (frame[5] & FLAG_DELTA_SYMBOL_DICT) == 0) {
-                return;
-            }
-            int[] pos = {12}; // just past the 12-byte QWP header
-            int deltaStart = readVarint(frame, pos);
-            int deltaCount = readVarint(frame, pos);
-            while (dict.size() < deltaStart) {
-                dict.add(null); // null-pad, mirroring the server
-            }
-            for (int i = 0; i < deltaCount; i++) {
-                int len = readVarint(frame, pos);
-                String symbol = new String(frame, pos[0], len, StandardCharsets.UTF_8);
-                pos[0] += len;
-                int idx = deltaStart + i;
-                while (dict.size() <= idx) {
-                    dict.add(null);
-                }
-                dict.set(idx, symbol);
-            }
-        }
-
-        private static byte[] buildAck(long seq) {
-            byte[] buf = new byte[1 + 8 + 2];
-            ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);
-            bb.put((byte) 0x00);
-            bb.putLong(seq);
-            bb.putShort((short) 0);
-            return buf;
-        }
-
-        private static int tableCount(byte[] frame) {
-            return (frame[6] & 0xFF) | ((frame[7] & 0xFF) << 8);
-        }
     }
 
     /**
@@ -523,7 +472,7 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                 if (newConnection) {
                     ackSequenceStarts.add(ackSequence);
                 }
-                client.sendBinary(CatchUpHandler.buildAck(ackSequence));
+                client.sendBinary(QwpWireTestUtils.buildAck(ackSequence));
                 if (connectionsAccepted.get() == 1) {
                     // Connection 1 registered the big symbol. Shrink the cap so the
                     // reconnect's catch-up budget can't fit it, then drop to force
@@ -585,16 +534,16 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
             }
             int connNumber = dictsByConn.size();
             List<String> dict = dictsByConn.get(connNumber - 1);
-            CatchUpHandler.accumulate(data, dict);
-            if (connNumber == 2 && CatchUpHandler.tableCount(data) == 0) {
+            QwpWireTestUtils.accumulateDeltaDictionary(data, dict);
+            if (connNumber == 2 && QwpWireTestUtils.tableCount(data) == 0) {
                 zeroTableFramesOnConn2++;
             }
             try {
                 long ackSequence = nextSeq.getAndIncrement();
                 if (newConnection) {
                     ackSequenceStarts.add(ackSequence);
                 }
-                client.sendBinary(CatchUpHandler.buildAck(ackSequence));
+                client.sendBinary(QwpWireTestUtils.buildAck(ackSequence));
                 // Drop connection 1 only once it has learned the entire batch, so
                 // the sender's sent-dictionary mirror is complete and the reconnect
                 // catch-up must replay a dictionary larger than the batch cap.

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/DeltaDictRecoveryTest.java

@@ -41,10 +41,8 @@
 import org.junit.Test;
 
 import java.io.IOException;
-import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
-import java.nio.charset.StandardCharsets;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -1099,20 +1097,25 @@ public void testRecoverySeedKeepsUtf8CollidingSymbolsInLockstep() throws Excepti
             }
             Assert.assertEquals("two colliding symbols must persist as two entries", 2, persistedSize);
 
-            // Phase 2: recover. The seeded producer id space must match pd.size().
+            // Phase 2: recover, then introduce a third symbol. The wire must extend
+            // the two recovered ids at id 2 rather than overwrite id 1.
             DictReconstructingHandler handler = new DictReconstructingHandler();
             try (TestWebSocketServer good = new TestWebSocketServer(handler)) {
                 int port = good.getPort();
                 good.start();
                 Assert.assertTrue(good.awaitStart(5, TimeUnit.SECONDS));
                 String cfg = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir + ";";
                 try (Sender s2 = Sender.fromConfig(cfg)) {
-                    Assert.assertEquals("recovered producer dictionary must match the persisted "
-                                    + "entry count (not de-duped), else the delta baseline desyncs from the mirror",
-                            persistedSize, globalDictSize(s2));
-                    Assert.assertEquals("delta baseline must resume at the persisted tip",
-                            persistedSize - 1, intField(s2, "sentMaxSymbolId"));
+                    s2.table("m").symbol("s", "tail").longColumn("v", 3L).atNow();
+                    long fsn = s2.flushAndGetSequence();
+                    Assert.assertTrue("the post-recovery frame must be acknowledged",
+                            s2.awaitAckedFsn(fsn, 5_000));
                 }
+                Assert.assertEquals("the new symbol delta must continue after both recovered entries",
+                        persistedSize, handler.lastDataDeltaStart());
+                Assert.assertEquals("the reconstructed dictionary must retain both colliding entries "
+                                + "and append the new symbol at id 2",
+                        Arrays.asList("?", "?", "tail"), handler.dictSnapshot());
             }
         });
     }
@@ -1369,19 +1372,6 @@ public void testFullDictFramesRecoverInFullDictModeInsteadOfBricking() throws Ex
         });
     }
 
-    private static int globalDictSize(Sender sender) throws Exception {
-        Field f = sender.getClass().getDeclaredField("globalSymbolDictionary");
-        f.setAccessible(true);
-        Object dict = f.get(sender);
-        return (int) dict.getClass().getMethod("size").invoke(dict);
-    }
-
-    private static int intField(Sender sender, String name) throws Exception {
-        Field f = sender.getClass().getDeclaredField(name);
-        f.setAccessible(true);
-        return f.getInt(sender);
-    }
-
     private static void writeAckWatermark(java.nio.file.Path path, long fsn) throws IOException {
         byte[] buf = new byte[16];
         ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);
@@ -1391,23 +1381,6 @@ private static void writeAckWatermark(java.nio.file.Path path, long fsn) throws
         java.nio.file.Files.write(path, buf);
     }
 
-    private static int readVarint(byte[] buf, int[] pos) {
-        int result = 0;
-        int shift = 0;
-        while (pos[0] < buf.length) {
-            int b = buf[pos[0]++] & 0xFF;
-            result |= (b & 0x7F) << shift;
-            if ((b & 0x80) == 0) {
-                return result;
-            }
-            shift += 7;
-            if (shift > 28) {
-                throw new IllegalStateException("varint too long");
-            }
-        }
-        throw new IllegalStateException("varint truncated");
-    }
-
     /**
      * Reconstructs the per-connection symbol dictionary from delta sections,
      * mirroring the server's {@code setQuick(deltaStart + i)} + null-padding.
@@ -1418,6 +1391,7 @@ private static class DictReconstructingHandler implements TestWebSocketServer.We
         private final List<String> dict = new ArrayList<>();
         private final AtomicLong nextSeq = new AtomicLong(0);
         private TestWebSocketServer.ClientHandler currentClient;
+        private volatile int lastDataDeltaStart = -1;
 
         synchronized List<Long> ackSequenceStarts() {
             return new ArrayList<>(ackSequenceStarts);
@@ -1431,6 +1405,10 @@ synchronized int maxDictSize() {
             return dict.size();
         }
 
+        int lastDataDeltaStart() {
+            return lastDataDeltaStart;
+        }
+
         @Override
         public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
             boolean newConnection = currentClient != client;
@@ -1439,59 +1417,23 @@ public synchronized void onBinaryMessage(TestWebSocketServer.ClientHandler clien
                 dict.clear(); // fresh server dictionary per connection
                 nextSeq.set(0);
             }
-            accumulate(data);
-            if (tableCount(data) == 0 && hasDelta(data)) {
+            QwpWireTestUtils.accumulateDeltaDictionary(data, dict);
+            int tableCount = QwpWireTestUtils.tableCount(data);
+            if (tableCount == 0 && QwpWireTestUtils.hasDelta(data)) {
                 sawCatchUpFrame = true;
+            } else if (tableCount > 0 && QwpWireTestUtils.hasDelta(data)) {
+                lastDataDeltaStart = QwpWireTestUtils.readVarint(data, new int[]{12});
             }
             try {
                 long ackSequence = nextSeq.getAndIncrement();
                 if (newConnection) {
                     ackSequenceStarts.add(ackSequence);
                 }
-                client.sendBinary(buildAck(ackSequence));
+                client.sendBinary(QwpWireTestUtils.buildAck(ackSequence));
             } catch (IOException e) {
                 throw new RuntimeException(e);
             }
         }
-
-        private static byte[] buildAck(long seq) {
-            byte[] buf = new byte[1 + 8 + 2];
-            ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);
-            bb.put((byte) 0x00);
-            bb.putLong(seq);
-            bb.putShort((short) 0);
-            return buf;
-        }
-
-        private static boolean hasDelta(byte[] frame) {
-            return frame.length >= 12 && (frame[5] & 0x08) != 0;
-        }
-
-        private static int tableCount(byte[] frame) {
-            return (frame[6] & 0xFF) | ((frame[7] & 0xFF) << 8);
-        }
-
-        private void accumulate(byte[] frame) {
-            if (!hasDelta(frame)) {
-                return;
-            }
-            int[] pos = {12};
-            int deltaStart = readVarint(frame, pos);
-            int deltaCount = readVarint(frame, pos);
-            while (dict.size() < deltaStart) {
-                dict.add(null);
-            }
-            for (int i = 0; i < deltaCount; i++) {
-                int len = readVarint(frame, pos);
-                String sym = new String(frame, pos[0], len, StandardCharsets.UTF_8);
-                pos[0] += len;
-                int idx = deltaStart + i;
-                while (dict.size() <= idx) {
-                    dict.add(null);
-                }
-                dict.set(idx, sym);
-            }
-        }
     }
 
     private static class SilentHandler implements TestWebSocketServer.WebSocketServerHandler {
@@ -1510,20 +1452,11 @@ private static class CountingHandler implements TestWebSocketServer.WebSocketSer
         public void onBinaryMessage(TestWebSocketServer.ClientHandler client, byte[] data) {
             frames.incrementAndGet();
             try {
-                client.sendBinary(buildAck(nextSeq.getAndIncrement()));
+                client.sendBinary(QwpWireTestUtils.buildAck(nextSeq.getAndIncrement()));
             } catch (IOException e) {
                 throw new RuntimeException(e);
             }
         }
-
-        private static byte[] buildAck(long seq) {
-            byte[] buf = new byte[1 + 8 + 2];
-            ByteBuffer bb = ByteBuffer.wrap(buf).order(ByteOrder.LITTLE_ENDIAN);
-            bb.put((byte) 0x00);
-            bb.putLong(seq);
-            bb.putShort((short) 0);
-            return buf;
-        }
     }
     /**
      * Short-writes the FIRST dictionary append after it is armed, modelling a disk that