Skip to content

Commit a55d3b9

Browse files
mtopolnikmtopolnik
committed
Fix duplicate column behavior
When user would add a value to the same column column twice for the current row, client would write that as the value for the next row, breaking the invariant that all cols have the same length. Fix is to align behavior with ILP: first write wins, others silently ignored.
1 parent 7fbfff5 commit a55d3b9

2 files changed

Lines changed: 92 additions & 30 deletions

File tree

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

Lines changed: 81 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -343,7 +343,9 @@ public QwpWebSocketSender boolColumn(CharSequence columnName, boolean value) {
343343
checkNotClosed();
344344
checkTableSelected();
345345
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_BOOLEAN, false);
346-
col.addBoolean(value);
346+
if (col != null) {
347+
col.addBoolean(value);
348+
}
347349
return this;
348350
}
349351

@@ -363,7 +365,9 @@ public QwpWebSocketSender byteColumn(CharSequence columnName, byte value) {
363365
checkNotClosed();
364366
checkTableSelected();
365367
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_BYTE, false);
366-
col.addByte(value);
368+
if (col != null) {
369+
col.addByte(value);
370+
}
367371
return this;
368372
}
369373

@@ -388,7 +392,9 @@ public QwpWebSocketSender charColumn(CharSequence columnName, char value) {
388392
checkNotClosed();
389393
checkTableSelected();
390394
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_CHAR, false);
391-
col.addShort((short) value);
395+
if (col != null) {
396+
col.addShort((short) value);
397+
}
392398
return this;
393399
}
394400

@@ -469,7 +475,9 @@ public Sender decimalColumn(CharSequence name, Decimal64 value) {
469475
checkNotClosed();
470476
checkTableSelected();
471477
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DECIMAL64, true);
472-
col.addDecimal64(value);
478+
if (col != null) {
479+
col.addDecimal64(value);
480+
}
473481
return this;
474482
}
475483

@@ -479,7 +487,9 @@ public Sender decimalColumn(CharSequence name, Decimal128 value) {
479487
checkNotClosed();
480488
checkTableSelected();
481489
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DECIMAL128, true);
482-
col.addDecimal128(value);
490+
if (col != null) {
491+
col.addDecimal128(value);
492+
}
483493
return this;
484494
}
485495

@@ -489,7 +499,9 @@ public Sender decimalColumn(CharSequence name, Decimal256 value) {
489499
checkNotClosed();
490500
checkTableSelected();
491501
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DECIMAL256, true);
492-
col.addDecimal256(value);
502+
if (col != null) {
503+
col.addDecimal256(value);
504+
}
493505
return this;
494506
}
495507

@@ -501,7 +513,9 @@ public Sender decimalColumn(CharSequence name, CharSequence value) {
501513
try {
502514
currentDecimal256.ofString(value);
503515
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DECIMAL256, true);
504-
col.addDecimal256(currentDecimal256);
516+
if (col != null) {
517+
col.addDecimal256(currentDecimal256);
518+
}
505519
} catch (Exception e) {
506520
throw new LineSenderException("Failed to parse decimal value: " + value, e);
507521
}
@@ -514,7 +528,9 @@ public Sender doubleArray(@NotNull CharSequence name, double[] values) {
514528
checkNotClosed();
515529
checkTableSelected();
516530
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DOUBLE_ARRAY, true);
517-
col.addDoubleArray(values);
531+
if (col != null) {
532+
col.addDoubleArray(values);
533+
}
518534
return this;
519535
}
520536

@@ -524,7 +540,9 @@ public Sender doubleArray(@NotNull CharSequence name, double[][] values) {
524540
checkNotClosed();
525541
checkTableSelected();
526542
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DOUBLE_ARRAY, true);
527-
col.addDoubleArray(values);
543+
if (col != null) {
544+
col.addDoubleArray(values);
545+
}
528546
return this;
529547
}
530548

@@ -534,7 +552,9 @@ public Sender doubleArray(@NotNull CharSequence name, double[][][] values) {
534552
checkNotClosed();
535553
checkTableSelected();
536554
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DOUBLE_ARRAY, true);
537-
col.addDoubleArray(values);
555+
if (col != null) {
556+
col.addDoubleArray(values);
557+
}
538558
return this;
539559
}
540560

@@ -544,7 +564,9 @@ public Sender doubleArray(CharSequence name, DoubleArray array) {
544564
checkNotClosed();
545565
checkTableSelected();
546566
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_DOUBLE_ARRAY, true);
547-
col.addDoubleArray(array);
567+
if (col != null) {
568+
col.addDoubleArray(array);
569+
}
548570
return this;
549571
}
550572

@@ -553,7 +575,9 @@ public QwpWebSocketSender doubleColumn(CharSequence columnName, double value) {
553575
checkNotClosed();
554576
checkTableSelected();
555577
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_DOUBLE, true);
556-
col.addDouble(value);
578+
if (col != null) {
579+
col.addDouble(value);
580+
}
557581
return this;
558582
}
559583

@@ -568,7 +592,9 @@ public QwpWebSocketSender floatColumn(CharSequence columnName, float value) {
568592
checkNotClosed();
569593
checkTableSelected();
570594
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_FLOAT, true);
571-
col.addFloat(value);
595+
if (col != null) {
596+
col.addFloat(value);
597+
}
572598
return this;
573599
}
574600

@@ -678,7 +704,9 @@ public QwpWebSocketSender intColumn(CharSequence columnName, int value) {
678704
checkNotClosed();
679705
checkTableSelected();
680706
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_INT, true);
681-
col.addInt(value);
707+
if (col != null) {
708+
col.addInt(value);
709+
}
682710
return this;
683711
}
684712

@@ -703,7 +731,9 @@ public QwpWebSocketSender long256Column(CharSequence columnName, long l0, long l
703731
checkNotClosed();
704732
checkTableSelected();
705733
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_LONG256, true);
706-
col.addLong256(l0, l1, l2, l3);
734+
if (col != null) {
735+
col.addLong256(l0, l1, l2, l3);
736+
}
707737
return this;
708738
}
709739

@@ -713,7 +743,9 @@ public Sender longArray(@NotNull CharSequence name, long[] values) {
713743
checkNotClosed();
714744
checkTableSelected();
715745
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_LONG_ARRAY, true);
716-
col.addLongArray(values);
746+
if (col != null) {
747+
col.addLongArray(values);
748+
}
717749
return this;
718750
}
719751

@@ -723,7 +755,9 @@ public Sender longArray(@NotNull CharSequence name, long[][] values) {
723755
checkNotClosed();
724756
checkTableSelected();
725757
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_LONG_ARRAY, true);
726-
col.addLongArray(values);
758+
if (col != null) {
759+
col.addLongArray(values);
760+
}
727761
return this;
728762
}
729763

@@ -733,7 +767,9 @@ public Sender longArray(@NotNull CharSequence name, long[][][] values) {
733767
checkNotClosed();
734768
checkTableSelected();
735769
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_LONG_ARRAY, true);
736-
col.addLongArray(values);
770+
if (col != null) {
771+
col.addLongArray(values);
772+
}
737773
return this;
738774
}
739775

@@ -743,7 +779,9 @@ public Sender longArray(@NotNull CharSequence name, LongArray array) {
743779
checkNotClosed();
744780
checkTableSelected();
745781
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(name), TYPE_LONG_ARRAY, true);
746-
col.addLongArray(array);
782+
if (col != null) {
783+
col.addLongArray(array);
784+
}
747785
return this;
748786
}
749787

@@ -752,7 +790,9 @@ public QwpWebSocketSender longColumn(CharSequence columnName, long value) {
752790
checkNotClosed();
753791
checkTableSelected();
754792
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_LONG, true);
755-
col.addLong(value);
793+
if (col != null) {
794+
col.addLong(value);
795+
}
756796
return this;
757797
}
758798

@@ -795,7 +835,9 @@ public QwpWebSocketSender shortColumn(CharSequence columnName, short value) {
795835
checkNotClosed();
796836
checkTableSelected();
797837
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_SHORT, false);
798-
col.addShort(value);
838+
if (col != null) {
839+
col.addShort(value);
840+
}
799841
return this;
800842
}
801843

@@ -804,7 +846,9 @@ public QwpWebSocketSender stringColumn(CharSequence columnName, CharSequence val
804846
checkNotClosed();
805847
checkTableSelected();
806848
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_STRING, true);
807-
col.addString(value);
849+
if (col != null) {
850+
col.addString(value);
851+
}
808852
return this;
809853
}
810854

@@ -813,7 +857,9 @@ public QwpWebSocketSender symbol(CharSequence columnName, CharSequence value) {
813857
checkNotClosed();
814858
checkTableSelected();
815859
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_SYMBOL, true);
816-
col.addSymbol(value);
860+
if (col != null) {
861+
col.addSymbol(value);
862+
}
817863
return this;
818864
}
819865

@@ -844,11 +890,15 @@ public QwpWebSocketSender timestampColumn(CharSequence columnName, long value, C
844890
checkTableSelected();
845891
if (unit == ChronoUnit.NANOS) {
846892
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_TIMESTAMP_NANOS, true);
847-
col.addLong(value);
893+
if (col != null) {
894+
col.addLong(value);
895+
}
848896
} else {
849897
long micros = toMicros(value, unit);
850898
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_TIMESTAMP, true);
851-
col.addLong(micros);
899+
if (col != null) {
900+
col.addLong(micros);
901+
}
852902
}
853903
return this;
854904
}
@@ -859,7 +909,9 @@ public QwpWebSocketSender timestampColumn(CharSequence columnName, Instant value
859909
checkTableSelected();
860910
long micros = value.getEpochSecond() * 1_000_000L + value.getNano() / 1000L;
861911
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_TIMESTAMP, true);
862-
col.addLong(micros);
912+
if (col != null) {
913+
col.addLong(micros);
914+
}
863915
return this;
864916
}
865917

@@ -875,7 +927,9 @@ public QwpWebSocketSender uuidColumn(CharSequence columnName, long lo, long hi)
875927
checkNotClosed();
876928
checkTableSelected();
877929
QwpTableBuffer.ColumnBuffer col = currentTableBuffer.getOrCreateColumn(checkedColumnName(columnName), TYPE_UUID, true);
878-
col.addUuid(hi, lo);
930+
if (col != null) {
931+
col.addUuid(hi, lo);
932+
}
879933
return this;
880934
}
881935

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

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,11 +32,11 @@
3232
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
3333
import io.questdb.client.std.CharSequenceIntHashMap;
3434
import io.questdb.client.std.Chars;
35-
import io.questdb.client.std.LowerCaseAsciiCharSequenceIntHashMap;
3635
import io.questdb.client.std.Decimal128;
3736
import io.questdb.client.std.Decimal256;
3837
import io.questdb.client.std.Decimal64;
3938
import io.questdb.client.std.Decimals;
39+
import io.questdb.client.std.LowerCaseAsciiCharSequenceIntHashMap;
4040
import io.questdb.client.std.MemoryTag;
4141
import io.questdb.client.std.NumericException;
4242
import io.questdb.client.std.ObjList;
@@ -196,13 +196,21 @@ public ColumnBuffer getExistingColumn(CharSequence name, byte type) {
196196
* <p>
197197
* Optimized for the common case where columns are accessed in the same
198198
* order every row: a sequential cursor avoids hash map lookups entirely.
199+
* <p>
200+
* Returns {@code null} when the column has already been written in the current
201+
* (uncommitted) row. Callers must treat a {@code null} return as "duplicate
202+
* column in this row — skip the write", matching the ILP first-value-wins
203+
* semantics. The check is a single field comparison on the hot path and has
204+
* no measurable cost.
199205
*/
200206
public ColumnBuffer getOrCreateColumn(CharSequence name, byte type, boolean nullable) {
201207
ColumnBuffer existing = lookupColumn(name, type);
202208
if (existing != null) {
203-
return existing;
209+
// col.size > rowCount means this column already received a value
210+
// for the in-progress row. Silently ignore the duplicate (first
211+
// value wins, same as the ILP server behaviour).
212+
return existing.size <= rowCount ? existing : null;
204213
}
205-
206214
return createColumn(name, type, nullable);
207215
}
208216

0 commit comments

Comments
 (0)