Skip to content

Commit a32a771

Browse files
jerrinotjerrinot
committed
optimize datagram size tracking and null padding
1 parent 34b4a11 commit a32a771

3 files changed

Lines changed: 101 additions & 20 deletions

File tree

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpUdpSender.java‎

Lines changed: 59 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,7 @@ public class QwpUdpSender implements Sender {
8484
private final UdpLineChannel channel;
8585
private final QwpColumnWriter columnWriter = new QwpColumnWriter();
8686
private final int maxDatagramSize;
87+
private final boolean trackDatagramEstimate;
8788
private final ObjList<ColumnEntry> rowJournal = new ObjList<>();
8889
private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
8990

@@ -96,6 +97,8 @@ public class QwpUdpSender implements Sender {
9697
private QwpTableBuffer currentTableBuffer;
9798
private String currentTableName;
9899
private int estimateColumnCount;
100+
private QwpTableBuffer.ColumnBuffer[] missingColumns = new QwpTableBuffer.ColumnBuffer[8];
101+
private int missingColumnCount;
99102
private int rowJournalSize;
100103
private long runningEstimate;
101104

@@ -107,6 +110,7 @@ public QwpUdpSender(NetworkFacade nf, int interfaceIPv4, int sendToAddress, int
107110
this.channel = new UdpLineChannel(nf, interfaceIPv4, sendToAddress, port, ttl);
108111
this.tableBuffers = new CharSequenceObjHashMap<>();
109112
this.maxDatagramSize = maxDatagramSize;
113+
this.trackDatagramEstimate = maxDatagramSize > 0;
110114
}
111115

112116
@Override
@@ -374,7 +378,7 @@ public Sender table(CharSequence tableName) {
374378
return this;
375379
}
376380
ensureNoInProgressRow("switch tables");
377-
if (maxDatagramSize > 0 && currentTableBuffer != null && currentTableBuffer.getRowCount() > 0) {
381+
if (trackDatagramEstimate && currentTableBuffer != null && currentTableBuffer.getRowCount() > 0) {
378382
flushSingleTable(currentTableName, currentTableBuffer);
379383
}
380384
cachedTimestampColumn = null;
@@ -440,7 +444,7 @@ private void appendBooleanColumn(CharSequence name, boolean value, boolean addJo
440444
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
441445
currentRowColumnCount++;
442446

443-
if (addJournal && maxDatagramSize > 0) {
447+
if (addJournal && trackDatagramEstimate) {
444448
ColumnEntry e = nextJournalEntry();
445449
e.kind = ENTRY_BOOL;
446450
e.name = col.getName();
@@ -458,7 +462,7 @@ private void appendDecimal128Column(CharSequence name, Decimal128 value, boolean
458462
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
459463
currentRowColumnCount++;
460464

461-
if (addJournal && maxDatagramSize > 0) {
465+
if (addJournal && trackDatagramEstimate) {
462466
ColumnEntry e = nextJournalEntry();
463467
e.kind = ENTRY_DECIMAL128;
464468
e.name = col.getName();
@@ -476,7 +480,7 @@ private void appendDecimal256Column(CharSequence name, Decimal256 value, boolean
476480
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
477481
currentRowColumnCount++;
478482

479-
if (addJournal && maxDatagramSize > 0) {
483+
if (addJournal && trackDatagramEstimate) {
480484
ColumnEntry e = nextJournalEntry();
481485
e.kind = ENTRY_DECIMAL256;
482486
e.name = col.getName();
@@ -494,7 +498,7 @@ private void appendDecimal64Column(CharSequence name, Decimal64 value, boolean a
494498
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
495499
currentRowColumnCount++;
496500

497-
if (addJournal && maxDatagramSize > 0) {
501+
if (addJournal && trackDatagramEstimate) {
498502
ColumnEntry e = nextJournalEntry();
499503
e.kind = ENTRY_DECIMAL64;
500504
e.name = col.getName();
@@ -527,7 +531,7 @@ private void appendDesignatedTimestamp(long value, boolean nanos, boolean addJou
527531
long payloadDelta = (long) (col.getValueCount() - valueCountBefore) * 8;
528532
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
529533

530-
if (addJournal && maxDatagramSize > 0) {
534+
if (addJournal && trackDatagramEstimate) {
531535
ColumnEntry e = nextJournalEntry();
532536
e.kind = nanos ? ENTRY_AT_NANOS : ENTRY_AT_MICROS;
533537
e.longValue = value;
@@ -563,7 +567,7 @@ private void appendDoubleArrayColumn(CharSequence name, Object value, boolean ad
563567
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
564568
currentRowColumnCount++;
565569

566-
if (addJournal && maxDatagramSize > 0) {
570+
if (addJournal && trackDatagramEstimate) {
567571
ColumnEntry e = nextJournalEntry();
568572
e.kind = ENTRY_DOUBLE_ARRAY;
569573
e.name = col.getName();
@@ -581,7 +585,7 @@ private void appendDoubleColumn(CharSequence name, double value, boolean addJour
581585
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
582586
currentRowColumnCount++;
583587

584-
if (addJournal && maxDatagramSize > 0) {
588+
if (addJournal && trackDatagramEstimate) {
585589
ColumnEntry e = nextJournalEntry();
586590
e.kind = ENTRY_DOUBLE;
587591
e.name = col.getName();
@@ -618,7 +622,7 @@ private void appendLongArrayColumn(CharSequence name, Object value, boolean addJ
618622
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
619623
currentRowColumnCount++;
620624

621-
if (addJournal && maxDatagramSize > 0) {
625+
if (addJournal && trackDatagramEstimate) {
622626
ColumnEntry e = nextJournalEntry();
623627
e.kind = ENTRY_LONG_ARRAY;
624628
e.name = col.getName();
@@ -636,7 +640,7 @@ private void appendLongColumn(CharSequence name, long value, boolean addJournal)
636640
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
637641
currentRowColumnCount++;
638642

639-
if (addJournal && maxDatagramSize > 0) {
643+
if (addJournal && trackDatagramEstimate) {
640644
ColumnEntry e = nextJournalEntry();
641645
e.kind = ENTRY_LONG;
642646
e.name = col.getName();
@@ -656,7 +660,7 @@ private void appendStringColumn(CharSequence name, CharSequence value, boolean a
656660
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
657661
currentRowColumnCount++;
658662

659-
if (addJournal && maxDatagramSize > 0) {
663+
if (addJournal && trackDatagramEstimate) {
660664
ColumnEntry e = nextJournalEntry();
661665
e.kind = ENTRY_STRING;
662666
e.name = col.getName();
@@ -675,7 +679,7 @@ private void appendSymbolColumn(CharSequence name, CharSequence value, boolean a
675679
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
676680
currentRowColumnCount++;
677681

678-
if (addJournal && maxDatagramSize > 0) {
682+
if (addJournal && trackDatagramEstimate) {
679683
ColumnEntry e = nextJournalEntry();
680684
e.kind = ENTRY_SYMBOL;
681685
e.name = col.getName();
@@ -693,15 +697,16 @@ private void appendTimestampColumn(CharSequence name, byte type, long value, byt
693697
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
694698
currentRowColumnCount++;
695699

696-
if (addJournal && maxDatagramSize > 0) {
700+
if (addJournal && trackDatagramEstimate) {
697701
ColumnEntry e = nextJournalEntry();
698702
e.kind = journalKind;
699703
e.name = col.getName();
700704
e.longValue = value;
701705
}
702706
}
703707

704-
private void applyRowPaddingEstimate(int targetRows) {
708+
private void collectMissingColumns(int targetRows) {
709+
missingColumnCount = 0;
705710
for (int i = 0, n = currentTableBuffer.getColumnCount(); i < n; i++) {
706711
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getColumn(i);
707712
int sizeBefore = col.getSize();
@@ -710,6 +715,13 @@ private void applyRowPaddingEstimate(int targetRows) {
710715
continue;
711716
}
712717

718+
ensureMissingColumnCapacity(missingColumnCount + 1);
719+
missingColumns[missingColumnCount++] = col;
720+
721+
if (!trackDatagramEstimate) {
722+
continue;
723+
}
724+
713725
if (col.isNullable()) {
714726
runningEstimate += bitmapBytes(sizeBefore + missing) - bitmapBytes(sizeBefore);
715727
continue;
@@ -721,6 +733,9 @@ private void applyRowPaddingEstimate(int targetRows) {
721733
}
722734

723735
private void applyValueEstimate(QwpTableBuffer.ColumnBuffer col, int sizeBefore, int sizeAfter, long payloadDelta) {
736+
if (!trackDatagramEstimate) {
737+
return;
738+
}
724739
runningEstimate += payloadDelta;
725740
if (col.isNullable()) {
726741
runningEstimate += bitmapBytes(sizeAfter) - bitmapBytes(sizeBefore);
@@ -761,16 +776,19 @@ private void commitCurrentRow() {
761776
}
762777

763778
int targetRows = currentTableBuffer.getRowCount() + 1;
764-
applyRowPaddingEstimate(targetRows);
779+
collectMissingColumns(targetRows);
765780

766-
if (maxDatagramSize > 0) {
781+
if (trackDatagramEstimate) {
767782
maybeAutoFlush();
768783
}
769784

770-
currentTableBuffer.nextRow();
771-
committedEstimate = runningEstimate;
772-
committedEstimateColumnCount = estimateColumnCount;
785+
currentTableBuffer.nextRow(missingColumns, missingColumnCount);
786+
if (trackDatagramEstimate) {
787+
committedEstimate = runningEstimate;
788+
committedEstimateColumnCount = estimateColumnCount;
789+
}
773790
currentRowColumnCount = 0;
791+
missingColumnCount = 0;
774792
rowJournalSize = 0;
775793
}
776794

@@ -818,6 +836,21 @@ private long estimateArrayValueSize(LongArray array) {
818836
return arraySizeCounter.size;
819837
}
820838

839+
private void ensureMissingColumnCapacity(int required) {
840+
if (required <= missingColumns.length) {
841+
return;
842+
}
843+
844+
int newCapacity = missingColumns.length;
845+
while (newCapacity < required) {
846+
newCapacity *= 2;
847+
}
848+
849+
QwpTableBuffer.ColumnBuffer[] newArr = new QwpTableBuffer.ColumnBuffer[newCapacity];
850+
System.arraycopy(missingColumns, 0, newArr, 0, missingColumnCount);
851+
missingColumns = newArr;
852+
}
853+
821854
private long estimateSymbolPayloadDelta(
822855
QwpTableBuffer.ColumnBuffer col,
823856
int valueCountBefore,
@@ -908,7 +941,7 @@ private void maybeAutoFlush() {
908941

909942
flushSingleTable(currentTableName, currentTableBuffer);
910943
replayRowJournal();
911-
applyRowPaddingEstimate(currentTableBuffer.getRowCount() + 1);
944+
collectMissingColumns(currentTableBuffer.getRowCount() + 1);
912945

913946
if (runningEstimate > maxDatagramSize) {
914947
throw singleRowTooLarge(runningEstimate);
@@ -983,6 +1016,7 @@ private void resetEstimateState() {
9831016
estimateColumnCount = 0;
9841017
committedEstimateColumnCount = 0;
9851018
currentRowColumnCount = 0;
1019+
missingColumnCount = 0;
9861020
}
9871021

9881022
private boolean hasInProgressRow() {
@@ -993,6 +1027,7 @@ private void rollbackEstimateToCommitted() {
9931027
runningEstimate = committedEstimate;
9941028
estimateColumnCount = committedEstimateColumnCount;
9951029
currentRowColumnCount = 0;
1030+
missingColumnCount = 0;
9961031
}
9971032

9981033
private LineSenderException singleRowTooLarge(long estimate) {
@@ -1003,6 +1038,10 @@ private LineSenderException singleRowTooLarge(long estimate) {
10031038
}
10041039

10051040
private void syncSchemaEstimate() {
1041+
if (!trackDatagramEstimate) {
1042+
return;
1043+
}
1044+
10061045
int newColumnCount = currentTableBuffer.getColumnCount();
10071046
if (newColumnCount == estimateColumnCount) {
10081047
return;

‎core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -300,6 +300,24 @@ public void nextRow() {
300300
committedColumnCount = columns.size();
301301
}
302302

303+
/**
304+
* Advances to the next row using a prepared list of columns that need null padding.
305+
* <p>
306+
* This avoids rescanning every column when the caller has already identified
307+
* which columns were omitted in the current row.
308+
*/
309+
public void nextRow(ColumnBuffer[] missingColumns, int missingColumnCount) {
310+
columnAccessCursor = 0;
311+
for (int i = 0; i < missingColumnCount; i++) {
312+
ColumnBuffer col = missingColumns[i];
313+
while (col.size < rowCount + 1) {
314+
col.addNull();
315+
}
316+
}
317+
rowCount++;
318+
committedColumnCount = columns.size();
319+
}
320+
303321
/**
304322
* Resets the buffer for reuse. Keeps column definitions and allocated memory.
305323
*/

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

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -292,6 +292,30 @@ public void testCancelRowTruncatesLateAddedColumnWhenSizeEqualsRowCount() throws
292292
});
293293
}
294294

295+
@Test
296+
public void testNextRowWithPreparedMissingColumnsPadsListedColumns() throws Exception {
297+
assertMemoryLeak(() -> {
298+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
299+
QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
300+
QwpTableBuffer.ColumnBuffer colB = table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
301+
302+
colA.addLong(10);
303+
colB.addString("x");
304+
table.nextRow();
305+
306+
colA.addLong(20);
307+
table.nextRow(new QwpTableBuffer.ColumnBuffer[]{colB}, 1);
308+
309+
assertEquals(2, colA.getSize());
310+
assertEquals(2, colA.getValueCount());
311+
assertEquals(2, colB.getSize());
312+
assertEquals(1, colB.getValueCount());
313+
assertFalse(colB.isNull(0));
314+
assertTrue(colB.isNull(1));
315+
}
316+
});
317+
}
318+
295319
@Test
296320
public void testCancelRowResetsDecimalScaleOnLateAddedColumn() throws Exception {
297321
assertMemoryLeak(() -> {

0 commit comments

Comments
 (0)