add tests for sender behavior with omitted columns and packet limits

Author: jerrinot

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

@@ -136,6 +136,74 @@ public void testBoundedSenderMixedNullablePaddingPreservesRowsAndPacketLimit() t
         });
     }
 
+    @Test
+    public void testUnboundedSenderOmittedNullableAndNonNullableColumnsPreservesRows() throws Exception {
+        assertMemoryLeak(() -> {
+            List<ScenarioRow> rows = Arrays.asList(
+                    row("t", sender -> sender.table("t")
+                                    .longColumn("x", 1)
+                                    .stringColumn("s", "alpha")
+                                    .symbol("sym", "one")
+                                    .atNow(),
+                            "x", 1L,
+                            "s", "alpha",
+                            "sym", "one"),
+                    row("t", sender -> sender.table("t")
+                                    .stringColumn("s", "beta")
+                                    .atNow(),
+                            "x", Long.MIN_VALUE,
+                            "s", "beta",
+                            "sym", null),
+                    row("t", sender -> sender.table("t")
+                                    .longColumn("x", 3)
+                                    .atNow(),
+                            "x", 3L,
+                            "s", null,
+                            "sym", null)
+            );
+
+            RunResult result = runScenario(rows, 0);
+
+            Assert.assertEquals(1, result.sendCount);
+            assertRowsEqual(expectedRows(rows), decodeRows(result.packets));
+        });
+    }
+
+    @Test
+    public void testBoundedSenderOmittedNonNullableColumnsPreservesRowsAndPacketLimit() throws Exception {
+        assertMemoryLeak(() -> {
+            String alpha = repeat('a', 256);
+            String beta = repeat('b', 192);
+            String omega = repeat('z', 256);
+            List<ScenarioRow> rows = Arrays.asList(
+                    row("mix", sender -> sender.table("mix")
+                                    .longColumn("x", 1)
+                                    .stringColumn("msg", alpha)
+                                    .atNow(),
+                            "x", 1L,
+                            "msg", alpha),
+                    row("mix", sender -> sender.table("mix")
+                                    .stringColumn("msg", beta)
+                                    .atNow(),
+                            "x", Long.MIN_VALUE,
+                            "msg", beta),
+                    row("mix", sender -> sender.table("mix")
+                                    .longColumn("x", 3)
+                                    .stringColumn("msg", omega)
+                                    .atNow(),
+                            "x", 3L,
+                            "msg", omega)
+            );
+
+            int maxDatagramSize = fullPacketSize(rows) - 1;
+            RunResult result = runScenario(rows, maxDatagramSize);
+
+            Assert.assertTrue("expected at least one auto-flush", result.packets.size() > 1);
+            assertPacketsWithinLimit(result, maxDatagramSize);
+            assertRowsEqual(expectedRows(rows), decodeRows(result.packets));
+        });
+    }
+
     @Test
     public void testBoundedSenderArrayReplayPreservesRowsAndPacketLimit() throws Exception {
         assertMemoryLeak(() -> {
@@ -552,6 +620,40 @@ public void testSchemaChangeAfterOutOfOrderExistingColumnsPreservesRows() throws
         });
     }
 
+    @Test
+    public void testBoundedSenderSchemaFlushThenOmittedNullableColumnsPreservesRows() throws Exception {
+        assertMemoryLeak(() -> {
+            List<ScenarioRow> rows = Arrays.asList(
+                    row("schema", sender -> sender.table("schema")
+                                    .longColumn("x", 1)
+                                    .stringColumn("s", "alpha")
+                                    .atNow(),
+                            "x", 1L,
+                            "s", "alpha"),
+                    row("schema", sender -> sender.table("schema")
+                                    .symbol("sym", "new")
+                                    .longColumn("x", 2)
+                                    .stringColumn("s", "beta")
+                                    .atNow(),
+                            "sym", "new",
+                            "x", 2L,
+                            "s", "beta"),
+                    row("schema", sender -> sender.table("schema")
+                                    .longColumn("x", 3)
+                                    .atNow(),
+                            "sym", null,
+                            "x", 3L,
+                            "s", null)
+            );
+
+            RunResult result = runScenario(rows, 1024 * 1024);
+
+            Assert.assertEquals(2, result.sendCount);
+            assertPacketsWithinLimit(result, 1024 * 1024);
+            assertRowsEqual(expectedRows(rows), decodeRows(result.packets));
+        });
+    }
+
     @Test
     public void testSymbolBoundary16383To16384PreservesRowsAndPacketLimit() throws Exception {
         assertMemoryLeak(() -> {

core/src/test/java/io/questdb/client/test/cutlass/qwp/protocol/QwpTableBufferTest.java

@@ -28,6 +28,7 @@
 import io.questdb.client.cutlass.line.array.DoubleArray;
 import io.questdb.client.cutlass.qwp.protocol.QwpConstants;
 import io.questdb.client.cutlass.qwp.protocol.QwpTableBuffer;
+import io.questdb.client.std.Unsafe;
 import io.questdb.client.std.Decimal128;
 import io.questdb.client.std.Decimal64;
 import org.junit.Test;
@@ -298,20 +299,26 @@ public void testNextRowWithPreparedMissingColumnsPadsListedColumns() throws Exce
             try (QwpTableBuffer table = new QwpTableBuffer("test")) {
                 QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
                 QwpTableBuffer.ColumnBuffer colB = table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
+                QwpTableBuffer.ColumnBuffer colC = table.getOrCreateColumn("c", QwpConstants.TYPE_LONG, false);
 
                 colA.addLong(10);
                 colB.addString("x");
+                colC.addLong(100);
                 table.nextRow();
 
                 colA.addLong(20);
-                table.nextRow(new QwpTableBuffer.ColumnBuffer[]{colB}, 1);
+                table.nextRow(new QwpTableBuffer.ColumnBuffer[]{colB, colC}, 2);
 
                 assertEquals(2, colA.getSize());
                 assertEquals(2, colA.getValueCount());
                 assertEquals(2, colB.getSize());
                 assertEquals(1, colB.getValueCount());
                 assertFalse(colB.isNull(0));
                 assertTrue(colB.isNull(1));
+                assertEquals(2, colC.getSize());
+                assertEquals(2, colC.getValueCount());
+                assertEquals(100L, Unsafe.getUnsafe().getLong(colC.getDataAddress()));
+                assertEquals(Long.MIN_VALUE, Unsafe.getUnsafe().getLong(colC.getDataAddress() + Long.BYTES));
             }
         });
     }