Skip to content

Commit 34b4a11

Browse files
jerrinotjerrinot
committed
optimize column lookup performance
1 parent 3de3e23 commit 34b4a11

4 files changed

Lines changed: 303 additions & 32 deletions

File tree

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

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -414,15 +414,18 @@ public Sender timestampColumn(CharSequence columnName, Instant value) {
414414
}
415415

416416
private QwpTableBuffer.ColumnBuffer acquireColumn(CharSequence name, byte type, boolean nullable) {
417-
boolean exists = currentTableBuffer.hasColumn(name);
418-
if (!exists && currentTableBuffer.getRowCount() > 0) {
417+
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getExistingColumn(name, type);
418+
if (col == null && currentTableBuffer.getRowCount() > 0) {
419419
if (currentRowColumnCount > 0) {
420420
throw new LineSenderException("schema change in middle of row is not supported");
421421
}
422422
flushSingleTable(currentTableName, currentTableBuffer);
423+
col = currentTableBuffer.getOrCreateColumn(name, type, nullable);
423424
}
424425

425-
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(name, type, nullable);
426+
if (col == null) {
427+
col = currentTableBuffer.getOrCreateColumn(name, type, nullable);
428+
}
426429
syncSchemaEstimate();
427430
return col;
428431
}

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

Lines changed: 46 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -149,8 +149,14 @@ public int getColumnCount() {
149149
return columns.size();
150150
}
151151

152-
public boolean hasColumn(CharSequence name) {
153-
return columnNameToIndex.get(name) != CharSequenceIntHashMap.NO_ENTRY_VALUE;
152+
/**
153+
* Returns an existing column with the given name and type, or {@code null} if absent.
154+
* <p>
155+
* Uses the same sequential access optimization as {@link #getOrCreateColumn(CharSequence, byte, boolean)}.
156+
* When the next expected column is accessed in order, the internal cursor advances without a hash lookup.
157+
*/
158+
public ColumnBuffer getExistingColumn(CharSequence name, byte type) {
159+
return lookupColumn(name, type);
154160
}
155161

156162
/**
@@ -175,36 +181,15 @@ public QwpColumnDef[] getColumnDefs() {
175181
* order every row: a sequential cursor avoids hash map lookups entirely.
176182
*/
177183
public ColumnBuffer getOrCreateColumn(CharSequence name, byte type, boolean nullable) {
178-
// Fast path: predict next column in sequence
179-
int n = columns.size();
180-
if (columnAccessCursor < n) {
181-
ColumnBuffer candidate = fastColumns[columnAccessCursor];
182-
if (Chars.equals(candidate.name, name)) {
183-
columnAccessCursor++;
184-
if (candidate.type != type) {
185-
throw new LineSenderException(
186-
"Column type mismatch for column '" + name + "': columnType="
187-
+ candidate.type + ", sentType=" + type
188-
);
189-
}
190-
return candidate;
191-
}
192-
}
193-
194-
// Slow path: hash map lookup
195-
int idx = columnNameToIndex.get(name);
196-
if (idx != CharSequenceIntHashMap.NO_ENTRY_VALUE) {
197-
ColumnBuffer existing = columns.get(idx);
198-
if (existing.type != type) {
199-
throw new LineSenderException(
200-
"Column type mismatch for column '" + name + "': columnType="
201-
+ existing.type + ", sentType=" + type
202-
);
203-
}
184+
ColumnBuffer existing = lookupColumn(name, type);
185+
if (existing != null) {
204186
return existing;
205187
}
206188

207-
// Create new column
189+
return createColumn(name, type, nullable);
190+
}
191+
192+
private ColumnBuffer createColumn(CharSequence name, byte type, boolean nullable) {
208193
ColumnBuffer col = new ColumnBuffer(Chars.toString(name), type, nullable);
209194
int index = columns.size();
210195
columns.add(col);
@@ -224,6 +209,38 @@ public ColumnBuffer getOrCreateColumn(CharSequence name, byte type, boolean null
224209
return col;
225210
}
226211

212+
private ColumnBuffer lookupColumn(CharSequence name, byte type) {
213+
// Fast path: predict next column in sequence
214+
int n = columns.size();
215+
if (columnAccessCursor < n) {
216+
ColumnBuffer candidate = fastColumns[columnAccessCursor];
217+
if (Chars.equals(candidate.name, name)) {
218+
columnAccessCursor++;
219+
assertColumnType(name, type, candidate);
220+
return candidate;
221+
}
222+
}
223+
224+
// Slow path: hash map lookup
225+
int idx = columnNameToIndex.get(name);
226+
if (idx != CharSequenceIntHashMap.NO_ENTRY_VALUE) {
227+
ColumnBuffer existing = columns.get(idx);
228+
assertColumnType(name, type, existing);
229+
return existing;
230+
}
231+
232+
return null;
233+
}
234+
235+
private static void assertColumnType(CharSequence name, byte type, ColumnBuffer column) {
236+
if (column.type != type) {
237+
throw new LineSenderException(
238+
"Column type mismatch for column '" + name + "': columnType="
239+
+ column.type + ", sentType=" + type
240+
);
241+
}
242+
}
243+
227244
/**
228245
* Returns the number of rows buffered.
229246
*/

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

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,45 @@ public void testBoundedSenderMixedTypesPreservesRowsAndPacketLimit() throws Exce
260260
});
261261
}
262262

263+
@Test
264+
public void testBoundedSenderOutOfOrderExistingColumnsPreservesRowsAndPacketLimit() throws Exception {
265+
assertMemoryLeak(() -> {
266+
List<ScenarioRow> rows = Arrays.asList(
267+
row("order", sender -> sender.table("order")
268+
.longColumn("a", 1)
269+
.stringColumn("b", "x")
270+
.symbol("c", "alpha")
271+
.atNow(),
272+
"a", 1L,
273+
"b", "x",
274+
"c", "alpha"),
275+
row("order", sender -> sender.table("order")
276+
.symbol("c", "beta")
277+
.stringColumn("b", "y")
278+
.longColumn("a", 2)
279+
.atNow(),
280+
"a", 2L,
281+
"b", "y",
282+
"c", "beta"),
283+
row("order", sender -> sender.table("order")
284+
.stringColumn("b", "z")
285+
.longColumn("a", 3)
286+
.symbol("c", null)
287+
.atNow(),
288+
"a", 3L,
289+
"b", "z",
290+
"c", null)
291+
);
292+
293+
int maxDatagramSize = fullPacketSize(rows) - 1;
294+
RunResult result = runScenario(rows, maxDatagramSize);
295+
296+
Assert.assertTrue("expected at least one auto-flush", result.packets.size() > 1);
297+
assertPacketsWithinLimit(result, maxDatagramSize);
298+
assertRowsEqual(expectedRows(rows), decodeRows(result.packets));
299+
});
300+
}
301+
263302
@Test
264303
public void testMixingAtNowAndAtMicrosAfterCommittedRowsThrowsSchemaChange() throws Exception {
265304
assertMemoryLeak(() -> {
@@ -478,6 +517,41 @@ public void testSchemaChangeWithCommittedRowsFlushesImmediately() throws Excepti
478517
});
479518
}
480519

520+
@Test
521+
public void testSchemaChangeAfterOutOfOrderExistingColumnsPreservesRows() throws Exception {
522+
assertMemoryLeak(() -> {
523+
List<ScenarioRow> rows = Arrays.asList(
524+
row("schema", sender -> sender.table("schema")
525+
.longColumn("a", 1)
526+
.stringColumn("b", "x")
527+
.atNow(),
528+
"a", 1L,
529+
"b", "x"),
530+
row("schema", sender -> sender.table("schema")
531+
.symbol("c", "new")
532+
.stringColumn("b", "y")
533+
.longColumn("a", 2)
534+
.atNow(),
535+
"a", 2L,
536+
"b", "y",
537+
"c", "new"),
538+
row("schema", sender -> sender.table("schema")
539+
.symbol("c", "next")
540+
.longColumn("a", 3)
541+
.stringColumn("b", "z")
542+
.atNow(),
543+
"a", 3L,
544+
"b", "z",
545+
"c", "next")
546+
);
547+
548+
RunResult result = runScenario(rows, 1024 * 1024);
549+
550+
Assert.assertEquals(2, result.sendCount);
551+
assertRowsEqual(expectedRows(rows), decodeRows(result.packets));
552+
});
553+
}
554+
481555
@Test
482556
public void testSymbolBoundary16383To16384PreservesRowsAndPacketLimit() throws Exception {
483557
assertMemoryLeak(() -> {

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

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -701,6 +701,183 @@ public void testLongArrayShrinkingSize() throws Exception {
701701
});
702702
}
703703

704+
@Test
705+
public void testGetExistingColumnReturnsOrderedColumnsAcrossRows() throws Exception {
706+
assertMemoryLeak(() -> {
707+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
708+
QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
709+
QwpTableBuffer.ColumnBuffer colB = table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
710+
colA.addLong(1);
711+
colB.addString("x");
712+
table.nextRow();
713+
714+
QwpTableBuffer.ColumnBuffer existingA = table.getExistingColumn("a", QwpConstants.TYPE_LONG);
715+
QwpTableBuffer.ColumnBuffer existingB = table.getExistingColumn("b", QwpConstants.TYPE_STRING);
716+
717+
assertSame(colA, existingA);
718+
assertSame(colB, existingB);
719+
720+
existingA.addLong(2);
721+
existingB.addString("y");
722+
table.nextRow();
723+
724+
assertEquals(2, table.getRowCount());
725+
assertEquals(2, colA.getSize());
726+
assertEquals(2, colA.getValueCount());
727+
assertEquals(2, colB.getSize());
728+
assertEquals(2, colB.getValueCount());
729+
}
730+
});
731+
}
732+
733+
@Test
734+
public void testGetExistingColumnReturnsOutOfOrderColumns() throws Exception {
735+
assertMemoryLeak(() -> {
736+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
737+
QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
738+
QwpTableBuffer.ColumnBuffer colB = table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
739+
colA.addLong(1);
740+
colB.addString("x");
741+
table.nextRow();
742+
743+
QwpTableBuffer.ColumnBuffer existingB = table.getExistingColumn("b", QwpConstants.TYPE_STRING);
744+
QwpTableBuffer.ColumnBuffer existingA = table.getExistingColumn("a", QwpConstants.TYPE_LONG);
745+
746+
assertSame(colB, existingB);
747+
assertSame(colA, existingA);
748+
749+
existingB.addString("y");
750+
existingA.addLong(2);
751+
table.nextRow();
752+
753+
assertEquals(2, table.getRowCount());
754+
assertEquals(2, colA.getSize());
755+
assertEquals(2, colA.getValueCount());
756+
assertEquals(2, colB.getSize());
757+
assertEquals(2, colB.getValueCount());
758+
}
759+
});
760+
}
761+
762+
@Test
763+
public void testGetExistingColumnReturnsNullWithoutCreatingColumn() throws Exception {
764+
assertMemoryLeak(() -> {
765+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
766+
QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
767+
colA.addLong(1);
768+
table.nextRow();
769+
770+
assertNull(table.getExistingColumn("missing", QwpConstants.TYPE_STRING));
771+
assertEquals(1, table.getColumnCount());
772+
773+
QwpTableBuffer.ColumnBuffer colB = table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
774+
assertNotNull(colB);
775+
assertEquals(2, table.getColumnCount());
776+
}
777+
});
778+
}
779+
780+
@Test
781+
public void testGetExistingColumnTypeMismatchOnOrderedPathThrows() throws Exception {
782+
assertMemoryLeak(() -> {
783+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
784+
QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
785+
table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
786+
colA.addLong(1);
787+
table.nextRow();
788+
789+
try {
790+
table.getExistingColumn("a", QwpConstants.TYPE_STRING);
791+
fail("Expected LineSenderException for ordered-path type mismatch");
792+
} catch (LineSenderException e) {
793+
assertTrue(e.getMessage().contains("Column type mismatch"));
794+
assertTrue(e.getMessage().contains("column 'a'"));
795+
}
796+
}
797+
});
798+
}
799+
800+
@Test
801+
public void testGetExistingColumnTypeMismatchOnHashPathThrows() throws Exception {
802+
assertMemoryLeak(() -> {
803+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
804+
QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
805+
QwpTableBuffer.ColumnBuffer colB = table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
806+
colA.addLong(1);
807+
colB.addString("x");
808+
table.nextRow();
809+
810+
try {
811+
table.getExistingColumn("b", QwpConstants.TYPE_LONG);
812+
fail("Expected LineSenderException for hash-path type mismatch");
813+
} catch (LineSenderException e) {
814+
assertTrue(e.getMessage().contains("Column type mismatch"));
815+
assertTrue(e.getMessage().contains("column 'b'"));
816+
}
817+
}
818+
});
819+
}
820+
821+
@Test
822+
public void testGetExistingColumnWorksAfterReset() throws Exception {
823+
assertMemoryLeak(() -> {
824+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
825+
QwpTableBuffer.ColumnBuffer colA = table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false);
826+
QwpTableBuffer.ColumnBuffer colB = table.getOrCreateColumn("b", QwpConstants.TYPE_STRING, true);
827+
colA.addLong(1);
828+
colB.addString("x");
829+
table.nextRow();
830+
831+
table.reset();
832+
833+
QwpTableBuffer.ColumnBuffer existingA = table.getExistingColumn("a", QwpConstants.TYPE_LONG);
834+
QwpTableBuffer.ColumnBuffer existingB = table.getExistingColumn("b", QwpConstants.TYPE_STRING);
835+
836+
assertSame(colA, existingA);
837+
assertSame(colB, existingB);
838+
839+
existingA.addLong(2);
840+
existingB.addString("y");
841+
table.nextRow();
842+
843+
assertEquals(1, table.getRowCount());
844+
assertEquals(1, colA.getSize());
845+
assertEquals(1, colA.getValueCount());
846+
assertEquals(1, colB.getSize());
847+
assertEquals(1, colB.getValueCount());
848+
}
849+
});
850+
}
851+
852+
@Test
853+
public void testGetExistingColumnWorksForLateAddedColumnAfterCancelRow() throws Exception {
854+
assertMemoryLeak(() -> {
855+
try (QwpTableBuffer table = new QwpTableBuffer("test")) {
856+
table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false).addLong(1);
857+
table.nextRow();
858+
859+
table.getOrCreateColumn("a", QwpConstants.TYPE_LONG, false).addLong(2);
860+
QwpTableBuffer.ColumnBuffer late = table.getOrCreateColumn("late", QwpConstants.TYPE_STRING, true);
861+
late.addString("stale");
862+
table.cancelCurrentRow();
863+
864+
QwpTableBuffer.ColumnBuffer existingLate = table.getExistingColumn("late", QwpConstants.TYPE_STRING);
865+
assertSame(late, existingLate);
866+
assertEquals(0, existingLate.getSize());
867+
assertEquals(0, existingLate.getValueCount());
868+
869+
table.getExistingColumn("a", QwpConstants.TYPE_LONG).addLong(2);
870+
table.nextRow();
871+
872+
assertEquals(2, table.getRowCount());
873+
assertEquals(2, existingLate.getSize());
874+
assertEquals(0, existingLate.getValueCount());
875+
assertTrue(existingLate.isNull(0));
876+
assertTrue(existingLate.isNull(1));
877+
}
878+
});
879+
}
880+
704881
/**
705882
* Simulates the encoder's walk over array data — the same logic as
706883
* QwpWebSocketEncoder.writeDoubleArrayColumn(). Returns the flat

0 commit comments

Comments
 (0)