Skip to content

Commit c4feb7b

Browse files
glasstigerglasstiger
committed
empty array support
1 parent 8ad622f commit c4feb7b

File tree

4 files changed

+248
-12
lines changed

4 files changed

+248
-12
lines changed

‎src/buffer/bufferv2.ts

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ class SenderBufferV2 extends SenderBufferBase {
8787
this.writeArrayValues(arr[i] as unknown[], dimensions);
8888
}
8989
} else {
90-
const type = typeof arr[0];
90+
const type = arr[0] ? typeof arr[0] : null;
9191
switch (type) {
9292
case "number":
9393
for (let i = 0; i < arr.length; i++) {
@@ -97,6 +97,9 @@ class SenderBufferV2 extends SenderBufferBase {
9797
);
9898
}
9999
break;
100+
case null:
101+
// empty array
102+
break;
100103
default:
101104
throw new Error(`Unsupported array type [type=${type}]`);
102105
}
@@ -117,6 +120,9 @@ class SenderBufferV2 extends SenderBufferBase {
117120
case "string":
118121
// in case of string[] capacity check is done separately for each array element
119122
return 0;
123+
case null:
124+
// empty array
125+
return 0;
120126
default:
121127
throw new Error(`Unsupported array type [type=${type}]`);
122128
}

‎src/utils.ts

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -43,9 +43,6 @@ function timestampToNanos(timestamp: bigint, unit: TimestampUnit) {
4343
function getDimensions(data: unknown) {
4444
const dimensions: number[] = [];
4545
while (Array.isArray(data)) {
46-
if (data.length === 0) {
47-
throw new Error("Zero length array not supported");
48-
}
4946
dimensions.push(data.length);
5047
data = data[0];
5148
}
@@ -88,15 +85,17 @@ function validateArray(data: unknown[], dimensions: number[]): ArrayPrimitive {
8885
}
8986
} else {
9087
// leaf level, expecting primitives
91-
if (expectedType === null) {
88+
if (expectedType === null && array[0]) {
9289
expectedType = typeof array[0] as ArrayPrimitive;
9390
}
9491

9592
for (let i = 0; i < array.length; i++) {
9693
const currentType = typeof array[i] as ArrayPrimitive;
9794
if (currentType !== expectedType) {
9895
throw new Error(
99-
`Mixed types found [expected=${expectedType}, current=${currentType}, path=${path}[${i}]]`,
96+
expectedType !== null
97+
? `Mixed types found [expected=${expectedType}, current=${currentType}, path=${path}[${i}]]`
98+
: `Unsupported array type [type=${currentType}]`,
10099
);
101100
}
102101
}

‎test/sender.buffer.test.ts

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -218,19 +218,39 @@ describe("Sender message builder test suite (anything not covered in client inte
218218
await sender.close();
219219
});
220220

221-
it("does not accept empty array", async function () {
221+
it("accepts empty array", async function () {
222222
const sender = new Sender({
223223
protocol: "tcp",
224224
protocol_version: "2",
225225
host: "host",
226226
init_buf_size: 1024,
227227
});
228-
sender.table("tableName");
229-
expect(() => sender.arrayColumn("arrayCol", [])).toThrow(
230-
"Zero length array not supported",
228+
await sender.table("tableName").arrayColumn("arrayCol", []).atNow();
229+
expect(bufferContentHex(sender)).toBe(
230+
toHex("tableName arrayCol==") + " 0e 0a 01 00 00 00 00 " + toHex("\n"),
231231
);
232-
expect(() => sender.arrayColumn("arrayCol", [[], []])).toThrow(
233-
"Zero length array not supported",
232+
await sender.close();
233+
});
234+
235+
it("accepts multi dimensional empty array", async function () {
236+
const sender = new Sender({
237+
protocol: "tcp",
238+
protocol_version: "2",
239+
host: "host",
240+
init_buf_size: 1024,
241+
});
242+
await sender
243+
.table("tableName")
244+
.arrayColumn("arrayCol", [
245+
[[], []],
246+
[[], []],
247+
[[], []],
248+
])
249+
.atNow();
250+
expect(bufferContentHex(sender)).toBe(
251+
toHex("tableName arrayCol==") +
252+
" 0e 0a 03 03 00 00 00 02 00 00 00 00 00 00 00 " +
253+
toHex("\n"),
234254
);
235255
await sender.close();
236256
});

‎test/sender.integration.test.ts

Lines changed: 211 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -246,6 +246,217 @@ describe("Sender tests with containerized QuestDB instance", () => {
246246
await sender.close();
247247
});
248248

249+
it("can ingest data via HTTP with protocol v2", async () => {
250+
const tableName = "test_http_v2";
251+
const schema = [
252+
{ name: "location", type: "SYMBOL" },
253+
{ name: "temperatures", type: "ARRAY", elemType: "DOUBLE", dim: 1 },
254+
{ name: "timestamp", type: "TIMESTAMP" },
255+
];
256+
257+
const sender = await Sender.fromConfig(
258+
`http::addr=${container.getHost()}:${container.getMappedPort(QUESTDB_HTTP_PORT)};auto_flush_rows=1`,
259+
);
260+
261+
// ingest via client
262+
await sender
263+
.table(tableName)
264+
.symbol("location", "us")
265+
.arrayColumn("temperatures", [17.1, 17.7, 18.4])
266+
.at(1658484765000000000n, "ns");
267+
268+
// wait for the table
269+
await waitForTable(container, tableName);
270+
271+
// query table
272+
const select1Result = await runSelect(container, tableName, 1);
273+
expect(select1Result.query).toBe(tableName);
274+
expect(select1Result.count).toBe(1);
275+
expect(select1Result.columns).toStrictEqual(schema);
276+
expect(select1Result.dataset).toStrictEqual([
277+
["us", [17.1, 17.7, 18.4], "2022-07-22T10:12:45.000000Z"],
278+
]);
279+
280+
// ingest via client, add new columns
281+
await sender
282+
.table(tableName)
283+
.symbol("location", "us")
284+
.arrayColumn("temperatures", [17.36, 18.4, 19.6, 18.7])
285+
.at(1658484765000666000n, "ns");
286+
await sender
287+
.table(tableName)
288+
.symbol("location", "emea")
289+
.symbol("city", "london")
290+
.arrayColumn("temperatures", [18.5, 18.4, 19.2])
291+
.floatColumn("daily_avg_temp", 18.7)
292+
.at(1658484765001234000n, "ns");
293+
294+
// query table
295+
const select2Result = await runSelect(container, tableName, 3);
296+
expect(select2Result.query).toBe(tableName);
297+
expect(select2Result.count).toBe(3);
298+
expect(select2Result.columns).toStrictEqual([
299+
{ name: "location", type: "SYMBOL" },
300+
{ name: "temperatures", type: "ARRAY", elemType: "DOUBLE", dim: 1 },
301+
{ name: "timestamp", type: "TIMESTAMP" },
302+
{ name: "city", type: "SYMBOL" },
303+
{ name: "daily_avg_temp", type: "DOUBLE" },
304+
]);
305+
expect(select2Result.dataset).toStrictEqual([
306+
["us", [17.1, 17.7, 18.4], "2022-07-22T10:12:45.000000Z", null, null],
307+
[
308+
"us",
309+
[17.36, 18.4, 19.6, 18.7],
310+
"2022-07-22T10:12:45.000666Z",
311+
null,
312+
null,
313+
],
314+
[
315+
"emea",
316+
[18.5, 18.4, 19.2],
317+
"2022-07-22T10:12:45.001234Z",
318+
"london",
319+
18.7,
320+
],
321+
]);
322+
323+
await sender.close();
324+
});
325+
326+
it("can ingest NULL array via HTTP with protocol v2", async () => {
327+
const tableName = "test_http_v2_null";
328+
const schema = [
329+
{ name: "location", type: "SYMBOL" },
330+
{ name: "temperatures", type: "ARRAY", elemType: "DOUBLE", dim: 1 },
331+
{ name: "timestamp", type: "TIMESTAMP" },
332+
];
333+
334+
const sender = await Sender.fromConfig(
335+
`http::addr=${container.getHost()}:${container.getMappedPort(QUESTDB_HTTP_PORT)}`,
336+
);
337+
338+
// ingest via client
339+
await sender
340+
.table(tableName)
341+
.symbol("location", "us")
342+
.arrayColumn("temperatures", [17.1, 17.7, 18.4])
343+
.at(1658484765000000000n, "ns");
344+
await sender
345+
.table(tableName)
346+
.symbol("location", "gb")
347+
.at(1658484765000666000n, "ns");
348+
await sender.flush();
349+
350+
// wait for the table
351+
await waitForTable(container, tableName);
352+
353+
// query table
354+
const select1Result = await runSelect(container, tableName, 2);
355+
expect(select1Result.query).toBe(tableName);
356+
expect(select1Result.count).toBe(2);
357+
expect(select1Result.columns).toStrictEqual(schema);
358+
expect(select1Result.dataset).toStrictEqual([
359+
["us", [17.1, 17.7, 18.4], "2022-07-22T10:12:45.000000Z"],
360+
["gb", null, "2022-07-22T10:12:45.000666Z"],
361+
]);
362+
363+
await sender.close();
364+
});
365+
366+
it("can ingest empty array via HTTP with protocol v2", async () => {
367+
const tableName = "test_http_v2_empty";
368+
const schema = [
369+
{ name: "location", type: "SYMBOL" },
370+
{ name: "temperatures", type: "ARRAY", elemType: "DOUBLE", dim: 1 },
371+
{ name: "timestamp", type: "TIMESTAMP" },
372+
];
373+
374+
const sender = await Sender.fromConfig(
375+
`http::addr=${container.getHost()}:${container.getMappedPort(QUESTDB_HTTP_PORT)}`,
376+
);
377+
378+
// ingest via client
379+
await sender
380+
.table(tableName)
381+
.symbol("location", "us")
382+
.arrayColumn("temperatures", [17.1, 17.7, 18.4])
383+
.at(1658484765000000000n, "ns");
384+
await sender
385+
.table(tableName)
386+
.symbol("location", "gb")
387+
.arrayColumn("temperatures", [])
388+
.at(1658484765000666000n, "ns");
389+
await sender.flush();
390+
391+
// wait for the table
392+
await waitForTable(container, tableName);
393+
394+
// query table
395+
const select1Result = await runSelect(container, tableName, 2);
396+
expect(select1Result.query).toBe(tableName);
397+
expect(select1Result.count).toBe(2);
398+
expect(select1Result.columns).toStrictEqual(schema);
399+
expect(select1Result.dataset).toStrictEqual([
400+
["us", [17.1, 17.7, 18.4], "2022-07-22T10:12:45.000000Z"],
401+
["gb", [], "2022-07-22T10:12:45.000666Z"],
402+
]);
403+
404+
await sender.close();
405+
});
406+
407+
it("can ingest multi dimensional empty array via HTTP with protocol v2", async () => {
408+
const tableName = "test_http_v2_multi_empty";
409+
const schema = [
410+
{ name: "location", type: "SYMBOL" },
411+
{ name: "temperatures", type: "ARRAY", elemType: "DOUBLE", dim: 2 },
412+
{ name: "timestamp", type: "TIMESTAMP" },
413+
];
414+
415+
const sender = await Sender.fromConfig(
416+
`http::addr=${container.getHost()}:${container.getMappedPort(QUESTDB_HTTP_PORT)}`,
417+
);
418+
419+
// ingest via client
420+
await sender
421+
.table(tableName)
422+
.symbol("location", "us")
423+
.arrayColumn("temperatures", [
424+
[17.1, 17.7],
425+
[18.4, 18.7],
426+
])
427+
.at(1658484765000000000n, "ns");
428+
await sender
429+
.table(tableName)
430+
.symbol("location", "gb")
431+
.arrayColumn("temperatures", [[], []])
432+
.at(1658484765000666000n, "ns");
433+
await sender.flush();
434+
435+
// wait for the table
436+
await waitForTable(container, tableName);
437+
438+
// query table
439+
const select1Result = await runSelect(container, tableName, 2);
440+
expect(select1Result.query).toBe(tableName);
441+
expect(select1Result.count).toBe(2);
442+
expect(select1Result.columns).toStrictEqual(schema);
443+
expect(select1Result.dataset).toStrictEqual([
444+
[
445+
"us",
446+
[
447+
[17.1, 17.7],
448+
[18.4, 18.7],
449+
],
450+
"2022-07-22T10:12:45.000000Z",
451+
],
452+
// todo: this should be [[], []]
453+
// probably a server bug
454+
["gb", [], "2022-07-22T10:12:45.000666Z"],
455+
]);
456+
457+
await sender.close();
458+
}, 60000000);
459+
249460
it("can ingest data via HTTP with auto flush interval", async () => {
250461
const tableName = "test_http_interval";
251462
const schema = [

0 commit comments

Comments
 (0)