Skip to content

Commit 716f0a3

Browse files
jerrinotjerrinot
committed
allow mid-row schema changes
1 parent 0cee86f commit 716f0a3

3 files changed

Lines changed: 241 additions & 50 deletions

File tree

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

Lines changed: 59 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,7 @@ public class QwpUdpSender implements Sender {
9999
private int estimateColumnCount;
100100
private QwpTableBuffer.ColumnBuffer[] missingColumns = new QwpTableBuffer.ColumnBuffer[8];
101101
private int missingColumnCount;
102+
private boolean replayingRowJournal;
102103
private int rowJournalSize;
103104
private long runningEstimate;
104105

@@ -158,6 +159,7 @@ public void cancelRow() {
158159
checkNotClosed();
159160
if (currentTableBuffer != null) {
160161
currentTableBuffer.cancelCurrentRow();
162+
currentTableBuffer.rollbackUncommittedColumns();
161163
rollbackEstimateToCommitted();
162164
}
163165
rowJournalSize = 0;
@@ -169,6 +171,7 @@ public void close() {
169171
try {
170172
if (hasInProgressRow()) {
171173
currentTableBuffer.cancelCurrentRow();
174+
currentTableBuffer.rollbackUncommittedColumns();
172175
rollbackEstimateToCommitted();
173176
rowJournalSize = 0;
174177
}
@@ -419,12 +422,13 @@ public Sender timestampColumn(CharSequence columnName, Instant value) {
419422

420423
private QwpTableBuffer.ColumnBuffer acquireColumn(CharSequence name, byte type, boolean nullable) {
421424
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getExistingColumn(name, type);
422-
if (col == null && currentTableBuffer.getRowCount() > 0) {
425+
assert !replayingRowJournal || currentTableBuffer.getRowCount() == 0;
426+
if (col == null && !replayingRowJournal && currentTableBuffer.getRowCount() > 0) {
423427
if (currentRowColumnCount > 0) {
424-
throw new LineSenderException("schema change in middle of row is not supported");
428+
flushCommittedRowsAndReplayCurrentRow();
429+
} else {
430+
flushSingleTable(currentTableName, currentTableBuffer);
425431
}
426-
flushSingleTable(currentTableName, currentTableBuffer);
427-
col = currentTableBuffer.getOrCreateColumn(name, type, nullable);
428432
}
429433

430434
if (col == null) {
@@ -444,7 +448,7 @@ private void appendBooleanColumn(CharSequence name, boolean value, boolean addJo
444448
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
445449
currentRowColumnCount++;
446450

447-
if (addJournal && trackDatagramEstimate) {
451+
if (shouldJournal(addJournal)) {
448452
ColumnEntry e = nextJournalEntry();
449453
e.kind = ENTRY_BOOL;
450454
e.name = col.getName();
@@ -462,7 +466,7 @@ private void appendDecimal128Column(CharSequence name, Decimal128 value, boolean
462466
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
463467
currentRowColumnCount++;
464468

465-
if (addJournal && trackDatagramEstimate) {
469+
if (shouldJournal(addJournal)) {
466470
ColumnEntry e = nextJournalEntry();
467471
e.kind = ENTRY_DECIMAL128;
468472
e.name = col.getName();
@@ -480,7 +484,7 @@ private void appendDecimal256Column(CharSequence name, Decimal256 value, boolean
480484
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
481485
currentRowColumnCount++;
482486

483-
if (addJournal && trackDatagramEstimate) {
487+
if (shouldJournal(addJournal)) {
484488
ColumnEntry e = nextJournalEntry();
485489
e.kind = ENTRY_DECIMAL256;
486490
e.name = col.getName();
@@ -498,7 +502,7 @@ private void appendDecimal64Column(CharSequence name, Decimal64 value, boolean a
498502
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
499503
currentRowColumnCount++;
500504

501-
if (addJournal && trackDatagramEstimate) {
505+
if (shouldJournal(addJournal)) {
502506
ColumnEntry e = nextJournalEntry();
503507
e.kind = ENTRY_DECIMAL64;
504508
e.name = col.getName();
@@ -531,7 +535,7 @@ private void appendDesignatedTimestamp(long value, boolean nanos, boolean addJou
531535
long payloadDelta = (long) (col.getValueCount() - valueCountBefore) * 8;
532536
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
533537

534-
if (addJournal && trackDatagramEstimate) {
538+
if (shouldJournal(addJournal)) {
535539
ColumnEntry e = nextJournalEntry();
536540
e.kind = nanos ? ENTRY_AT_NANOS : ENTRY_AT_MICROS;
537541
e.longValue = value;
@@ -567,7 +571,7 @@ private void appendDoubleArrayColumn(CharSequence name, Object value, boolean ad
567571
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
568572
currentRowColumnCount++;
569573

570-
if (addJournal && trackDatagramEstimate) {
574+
if (shouldJournal(addJournal)) {
571575
ColumnEntry e = nextJournalEntry();
572576
e.kind = ENTRY_DOUBLE_ARRAY;
573577
e.name = col.getName();
@@ -585,7 +589,7 @@ private void appendDoubleColumn(CharSequence name, double value, boolean addJour
585589
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
586590
currentRowColumnCount++;
587591

588-
if (addJournal && trackDatagramEstimate) {
592+
if (shouldJournal(addJournal)) {
589593
ColumnEntry e = nextJournalEntry();
590594
e.kind = ENTRY_DOUBLE;
591595
e.name = col.getName();
@@ -622,7 +626,7 @@ private void appendLongArrayColumn(CharSequence name, Object value, boolean addJ
622626
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
623627
currentRowColumnCount++;
624628

625-
if (addJournal && trackDatagramEstimate) {
629+
if (shouldJournal(addJournal)) {
626630
ColumnEntry e = nextJournalEntry();
627631
e.kind = ENTRY_LONG_ARRAY;
628632
e.name = col.getName();
@@ -640,7 +644,7 @@ private void appendLongColumn(CharSequence name, long value, boolean addJournal)
640644
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
641645
currentRowColumnCount++;
642646

643-
if (addJournal && trackDatagramEstimate) {
647+
if (shouldJournal(addJournal)) {
644648
ColumnEntry e = nextJournalEntry();
645649
e.kind = ENTRY_LONG;
646650
e.name = col.getName();
@@ -660,7 +664,7 @@ private void appendStringColumn(CharSequence name, CharSequence value, boolean a
660664
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
661665
currentRowColumnCount++;
662666

663-
if (addJournal && trackDatagramEstimate) {
667+
if (shouldJournal(addJournal)) {
664668
ColumnEntry e = nextJournalEntry();
665669
e.kind = ENTRY_STRING;
666670
e.name = col.getName();
@@ -679,7 +683,7 @@ private void appendSymbolColumn(CharSequence name, CharSequence value, boolean a
679683
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
680684
currentRowColumnCount++;
681685

682-
if (addJournal && trackDatagramEstimate) {
686+
if (shouldJournal(addJournal)) {
683687
ColumnEntry e = nextJournalEntry();
684688
e.kind = ENTRY_SYMBOL;
685689
e.name = col.getName();
@@ -697,7 +701,7 @@ private void appendTimestampColumn(CharSequence name, byte type, long value, byt
697701
applyValueEstimate(col, sizeBefore, col.getSize(), payloadDelta);
698702
currentRowColumnCount++;
699703

700-
if (addJournal && trackDatagramEstimate) {
704+
if (shouldJournal(addJournal)) {
701705
ColumnEntry e = nextJournalEntry();
702706
e.kind = journalKind;
703707
e.name = col.getName();
@@ -927,6 +931,14 @@ private void flushSingleTable(String tableName, QwpTableBuffer tableBuffer) {
927931
resetEstimateState();
928932
}
929933

934+
private void flushCommittedRowsAndReplayCurrentRow() {
935+
currentTableBuffer.cancelCurrentRow();
936+
currentTableBuffer.rollbackUncommittedColumns();
937+
rollbackEstimateToCommitted();
938+
flushSingleTable(currentTableName, currentTableBuffer);
939+
replayRowJournal();
940+
}
941+
930942
private void maybeAutoFlush() {
931943
if (runningEstimate <= maxDatagramSize) {
932944
return;
@@ -936,11 +948,7 @@ private void maybeAutoFlush() {
936948
throw singleRowTooLarge(runningEstimate);
937949
}
938950

939-
currentTableBuffer.cancelCurrentRow();
940-
rollbackEstimateToCommitted();
941-
942-
flushSingleTable(currentTableName, currentTableBuffer);
943-
replayRowJournal();
951+
flushCommittedRowsAndReplayCurrentRow();
944952
collectMissingColumns(currentTableBuffer.getRowCount() + 1);
945953

946954
if (runningEstimate > maxDatagramSize) {
@@ -986,27 +994,33 @@ private ColumnEntry nextJournalEntry() {
986994
}
987995

988996
private void replayRowJournal() {
989-
for (int i = 0; i < rowJournalSize; i++) {
990-
ColumnEntry entry = rowJournal.getQuick(i);
991-
switch (entry.kind) {
992-
case ENTRY_AT_MICROS -> appendDesignatedTimestamp(entry.longValue, false, false);
993-
case ENTRY_AT_NANOS -> appendDesignatedTimestamp(entry.longValue, true, false);
994-
case ENTRY_BOOL -> appendBooleanColumn(entry.name, entry.boolValue, false);
995-
case ENTRY_DECIMAL128 -> appendDecimal128Column(entry.name, (Decimal128) entry.objectValue, false);
996-
case ENTRY_DECIMAL256 -> appendDecimal256Column(entry.name, (Decimal256) entry.objectValue, false);
997-
case ENTRY_DECIMAL64 -> appendDecimal64Column(entry.name, (Decimal64) entry.objectValue, false);
998-
case ENTRY_DOUBLE -> appendDoubleColumn(entry.name, entry.doubleValue, false);
999-
case ENTRY_DOUBLE_ARRAY -> appendDoubleArrayColumn(entry.name, entry.objectValue, false);
1000-
case ENTRY_LONG -> appendLongColumn(entry.name, entry.longValue, false);
1001-
case ENTRY_LONG_ARRAY -> appendLongArrayColumn(entry.name, entry.objectValue, false);
1002-
case ENTRY_STRING -> appendStringColumn(entry.name, entry.stringValue, false);
1003-
case ENTRY_SYMBOL -> appendSymbolColumn(entry.name, entry.stringValue, false);
1004-
case ENTRY_TIMESTAMP_COL_MICROS ->
1005-
appendTimestampColumn(entry.name, TYPE_TIMESTAMP, entry.longValue, ENTRY_TIMESTAMP_COL_MICROS, false);
1006-
case ENTRY_TIMESTAMP_COL_NANOS ->
1007-
appendTimestampColumn(entry.name, TYPE_TIMESTAMP_NANOS, entry.longValue, ENTRY_TIMESTAMP_COL_NANOS, false);
1008-
default -> throw new LineSenderException("unknown row journal entry type: " + entry.kind);
997+
assert currentTableBuffer.getRowCount() == 0 : "row journal replay requires a reset table buffer";
998+
replayingRowJournal = true;
999+
try {
1000+
for (int i = 0; i < rowJournalSize; i++) {
1001+
ColumnEntry entry = rowJournal.getQuick(i);
1002+
switch (entry.kind) {
1003+
case ENTRY_AT_MICROS -> appendDesignatedTimestamp(entry.longValue, false, false);
1004+
case ENTRY_AT_NANOS -> appendDesignatedTimestamp(entry.longValue, true, false);
1005+
case ENTRY_BOOL -> appendBooleanColumn(entry.name, entry.boolValue, false);
1006+
case ENTRY_DECIMAL128 -> appendDecimal128Column(entry.name, (Decimal128) entry.objectValue, false);
1007+
case ENTRY_DECIMAL256 -> appendDecimal256Column(entry.name, (Decimal256) entry.objectValue, false);
1008+
case ENTRY_DECIMAL64 -> appendDecimal64Column(entry.name, (Decimal64) entry.objectValue, false);
1009+
case ENTRY_DOUBLE -> appendDoubleColumn(entry.name, entry.doubleValue, false);
1010+
case ENTRY_DOUBLE_ARRAY -> appendDoubleArrayColumn(entry.name, entry.objectValue, false);
1011+
case ENTRY_LONG -> appendLongColumn(entry.name, entry.longValue, false);
1012+
case ENTRY_LONG_ARRAY -> appendLongArrayColumn(entry.name, entry.objectValue, false);
1013+
case ENTRY_STRING -> appendStringColumn(entry.name, entry.stringValue, false);
1014+
case ENTRY_SYMBOL -> appendSymbolColumn(entry.name, entry.stringValue, false);
1015+
case ENTRY_TIMESTAMP_COL_MICROS ->
1016+
appendTimestampColumn(entry.name, TYPE_TIMESTAMP, entry.longValue, ENTRY_TIMESTAMP_COL_MICROS, false);
1017+
case ENTRY_TIMESTAMP_COL_NANOS ->
1018+
appendTimestampColumn(entry.name, TYPE_TIMESTAMP_NANOS, entry.longValue, ENTRY_TIMESTAMP_COL_NANOS, false);
1019+
default -> throw new LineSenderException("unknown row journal entry type: " + entry.kind);
1020+
}
10091021
}
1022+
} finally {
1023+
replayingRowJournal = false;
10101024
}
10111025
}
10121026

@@ -1030,6 +1044,10 @@ private void rollbackEstimateToCommitted() {
10301044
missingColumnCount = 0;
10311045
}
10321046

1047+
private boolean shouldJournal(boolean addJournal) {
1048+
return addJournal && (trackDatagramEstimate || currentTableBuffer.getRowCount() > 0 || rowJournalSize > 0);
1049+
}
1050+
10331051
private LineSenderException singleRowTooLarge(long estimate) {
10341052
return new LineSenderException(
10351053
"single row exceeds maximum datagram size (" + maxDatagramSize

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

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,43 @@ private ColumnBuffer createColumn(CharSequence name, byte type, boolean nullable
209209
return col;
210210
}
211211

212+
public void rollbackUncommittedColumns() {
213+
if (columns.size() <= committedColumnCount) {
214+
return;
215+
}
216+
217+
for (int i = columns.size() - 1; i >= committedColumnCount; i--) {
218+
ColumnBuffer col = columns.getQuick(i);
219+
if (col != null) {
220+
col.close();
221+
}
222+
columns.remove(i);
223+
}
224+
rebuildColumnAccessStructures();
225+
}
226+
227+
private void rebuildColumnAccessStructures() {
228+
columnNameToIndex.clear();
229+
230+
int columnCount = columns.size();
231+
int minCapacity = Math.max(8, columnCount + 4);
232+
if (fastColumns == null || fastColumns.length < minCapacity) {
233+
fastColumns = new ColumnBuffer[minCapacity];
234+
} else {
235+
Arrays.fill(fastColumns, null);
236+
}
237+
238+
for (int i = 0; i < columnCount; i++) {
239+
ColumnBuffer col = columns.getQuick(i);
240+
fastColumns[i] = col;
241+
columnNameToIndex.put(col.name, i);
242+
}
243+
244+
schemaHashComputed = false;
245+
columnDefsCacheValid = false;
246+
cachedColumnDefs = null;
247+
}
248+
212249
private ColumnBuffer lookupColumn(CharSequence name, byte type) {
213250
// Fast path: predict next column in sequence
214251
int n = columns.size();

0 commit comments

Comments
 (0)