diff --git a/src/buffer/base.ts b/src/buffer/base.ts index dd31423..2e61ddb 100644 --- a/src/buffer/base.ts +++ b/src/buffer/base.ts @@ -9,12 +9,7 @@ import { DEFAULT_BUFFER_SIZE, DEFAULT_MAX_BUFFER_SIZE, } from "./index"; -import { - isInteger, - timestampToMicros, - timestampToNanos, - TimestampUnit, -} from "../utils"; +import { isInteger, TimestampUnit } from "../utils"; // Default maximum length for table and column names. const DEFAULT_MAX_NAME_LENGTH = 127; @@ -269,6 +264,12 @@ abstract class SenderBufferBase implements SenderBuffer { return this; } + protected abstract writeTimestamp( + timestamp: number | bigint, + unit: TimestampUnit, + designated: boolean, + ): void; + /** * Writes a timestamp column with its value into the buffer.
* Use it to insert into TIMESTAMP columns. @@ -284,15 +285,18 @@ abstract class SenderBufferBase implements SenderBuffer { unit: TimestampUnit = "us", ): SenderBuffer { if (typeof value !== "bigint" && !Number.isInteger(value)) { - throw new Error(`Value must be an integer or BigInt, received ${value}`); + throw new Error( + `Timestamp value must be an integer or BigInt, received ${value}`, + ); } - this.writeColumn(name, value, () => { - const valueMicros = timestampToMicros(BigInt(value), unit); - const valueStr = valueMicros.toString(); - this.checkCapacity([valueStr], 1); - this.write(valueStr); - this.write("t"); - }); + if (unit == "ns" && typeof value !== "bigint") { + throw new Error( + `Timestamp value must be a BigInt if it is set in nanoseconds`, + ); + } + this.writeColumn(name, value, () => + this.writeTimestamp(value, unit, false), + ); return this; } @@ -313,11 +317,14 @@ abstract class SenderBufferBase implements SenderBuffer { `Designated timestamp must be an integer or BigInt, received ${timestamp}`, ); } - const timestampNanos = timestampToNanos(BigInt(timestamp), unit); - const timestampStr = timestampNanos.toString(); - this.checkCapacity([timestampStr], 2); + if (unit == "ns" && typeof timestamp !== "bigint") { + throw new Error( + `Designated timestamp must be a BigInt if it is set in nanoseconds`, + ); + } + this.checkCapacity([], 1); this.write(" "); - this.write(timestampStr); + this.writeTimestamp(timestamp, unit, true); this.write("\n"); this.startNewRow(); } diff --git a/src/buffer/bufferv1.ts b/src/buffer/bufferv1.ts index 64cfae2..05442e0 100644 --- a/src/buffer/bufferv1.ts +++ b/src/buffer/bufferv1.ts @@ -2,6 +2,7 @@ import { SenderOptions } from "../options"; import { SenderBuffer } from "./index"; import { SenderBufferBase } from "./base"; +import { timestampToMicros, timestampToNanos, TimestampUnit } from "../utils"; /** * Buffer implementation for protocol version 1.
@@ -39,6 +40,23 @@ class SenderBufferV1 extends SenderBufferBase { return this; } + protected writeTimestamp( + timestamp: number | bigint, + unit: TimestampUnit = "us", + designated: boolean, + ): void { + const biValue = BigInt(timestamp); + const timestampValue = designated + ? timestampToNanos(biValue, unit) + : timestampToMicros(biValue, unit); + const timestampStr = timestampValue.toString(); + this.checkCapacity([timestampStr], 2); + this.write(timestampStr); + if (!designated) { + this.write("t"); + } + } + /** * Array columns are not supported in protocol v1. * diff --git a/src/buffer/bufferv2.ts b/src/buffer/bufferv2.ts index 711a57a..b24c176 100644 --- a/src/buffer/bufferv2.ts +++ b/src/buffer/bufferv2.ts @@ -2,7 +2,13 @@ import { SenderOptions } from "../options"; import { SenderBuffer } from "./index"; import { SenderBufferBase } from "./base"; -import { ArrayPrimitive, getDimensions, validateArray } from "../utils"; +import { + ArrayPrimitive, + getDimensions, + timestampToMicros, + TimestampUnit, + validateArray, +} from "../utils"; // Column type constants for protocol v2. const COLUMN_TYPE_DOUBLE: number = 10; @@ -53,6 +59,28 @@ class SenderBufferV2 extends SenderBufferBase { return this; } + protected writeTimestamp( + value: number | bigint, + unit: TimestampUnit = "us", + ): void { + let biValue: bigint; + let suffix: string; + switch (unit) { + case "ns": + biValue = BigInt(value); + suffix = "n"; + break; + default: + biValue = timestampToMicros(BigInt(value), unit); + suffix = "t"; + } + + const timestampStr = biValue.toString(); + this.checkCapacity([timestampStr], 2); + this.write(timestampStr); + this.write(suffix); + } + /** * Write an array column with its values into the buffer using v2 format. * diff --git a/src/options.ts b/src/options.ts index 06b64a4..8732c01 100644 --- a/src/options.ts +++ b/src/options.ts @@ -226,7 +226,7 @@ class SenderOptions { * If TCP transport is used, the protocol version will default to 1. * In case of HTTP transport the /settings endpoint of the database is used to find the protocol versions * supported by the server, and the highest will be selected. - * When calling the /settings endpoint the timeout and TLs options are used from the options object. + * When calling the /settings endpoint the timeout and TLS options are used from the options object. * @param {SenderOptions} options SenderOptions instance needs resolving protocol version */ static async resolveAuto(options: SenderOptions) { diff --git a/test/logging.test.ts b/test/logging.test.ts index 6e17e34..fecd353 100644 --- a/test/logging.test.ts +++ b/test/logging.test.ts @@ -9,7 +9,7 @@ import { vi, } from "vitest"; -import { Logger } from "../src/logging"; +import { Logger } from "../src"; describe("Default logging suite", function () { const error = vi.spyOn(console, "error").mockImplementation(() => {}); diff --git a/test/options.test.ts b/test/options.test.ts index 226b005..fcbbab3 100644 --- a/test/options.test.ts +++ b/test/options.test.ts @@ -2,7 +2,7 @@ import { describe, it, expect, beforeAll, afterAll } from "vitest"; import { Agent } from "undici"; -import { SenderOptions } from "../src/options"; +import { SenderOptions } from "../src"; import { MockHttp } from "./util/mockhttp"; import { readFileSync } from "fs"; diff --git a/test/sender.buffer.test.ts b/test/sender.buffer.test.ts index f9f5119..f5f2406 100644 --- a/test/sender.buffer.test.ts +++ b/test/sender.buffer.test.ts @@ -2,8 +2,7 @@ import { describe, it, expect } from "vitest"; import { readFileSync } from "fs"; -import { Sender } from "../src"; -import { SenderOptions } from "../src/options"; +import { Sender, SenderOptions } from "../src"; describe("Client interop test suite", function () { it("runs client tests as per json test config", async function () { @@ -94,27 +93,6 @@ describe("Client interop test suite", function () { }); describe("Sender message builder test suite (anything not covered in client interop test suite)", function () { - it("throws on invalid timestamp unit", async function () { - const sender = new Sender({ - protocol: "tcp", - protocol_version: "1", - host: "host", - auto_flush: false, - init_buf_size: 1024, - }); - - await expect( - async () => - await sender - .table("tableName") - .booleanColumn("boolCol", true) - // @ts-expect-error - Testing invalid options - .timestampColumn("timestampCol", 1658484765000000, "foobar") - .atNow(), - ).rejects.toThrow("Unknown timestamp unit: foobar"); - await sender.close(); - }); - it("supports json object", async function () { const pages: Array<{ id: string; @@ -442,115 +420,104 @@ describe("Sender message builder test suite (anything not covered in client inte await sender.close(); }); - it("supports timestamp field as number", async function () { + it("throws on invalid timestamp unit", async function () { const sender = new Sender({ protocol: "tcp", protocol_version: "1", host: "host", + auto_flush: false, init_buf_size: 1024, }); - await sender - .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) - .atNow(); - expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t\n", - ); + + await expect( + async () => + await sender + .table("tableName") + .booleanColumn("boolCol", true) + // @ts-expect-error - Testing invalid timestamp unit + .timestampColumn("timestampCol", 1658484765000000, "foobar") + .atNow(), + ).rejects.toThrow("Unknown timestamp unit: foobar"); await sender.close(); }); - it("supports timestamp field as ns number", async function () { + it("supports timestamp field as number for 'us' and 'ms' units with protocol v1", async function () { const sender = new Sender({ protocol: "tcp", protocol_version: "1", host: "host", init_buf_size: 1024, }); + await expect( + async () => + await sender + .table("tableName") + .timestampColumn("ts", 1658484765000000, "ns") + .atNow(), + ).rejects.toThrow( + "Timestamp value must be a BigInt if it is set in nanoseconds", + ); + + sender.reset(); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000, "ns") + .timestampColumn("ts", 1658484765123456) .atNow(); - expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000t\n", - ); - await sender.close(); - }); - - it("supports timestamp field as us number", async function () { - const sender = new Sender({ - protocol: "tcp", - protocol_version: "1", - host: "host", - init_buf_size: 1024, - }); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000, "us") + .timestampColumn("ts", 1658484765123456, "us") .atNow(); - expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t\n", - ); - await sender.close(); - }); - - it("supports timestamp field as ms number", async function () { - const sender = new Sender({ - protocol: "tcp", - protocol_version: "1", - host: "host", - init_buf_size: 1024, - }); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000, "ms") + .timestampColumn("ts", 1658484765123, "ms") .atNow(); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t\n", + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123000t\n", ); await sender.close(); }); - it("supports timestamp field as BigInt", async function () { + it("supports timestamp field as number for 'us' and 'ms' units with protocol v2", async function () { const sender = new Sender({ protocol: "tcp", - protocol_version: "1", + protocol_version: "2", host: "host", init_buf_size: 1024, }); + await expect( + async () => + await sender + .table("tableName") + .timestampColumn("ts", 1658484765000000, "ns") + .atNow(), + ).rejects.toThrow( + "Timestamp value must be a BigInt if it is set in nanoseconds", + ); + + sender.reset(); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000n) + .timestampColumn("ts", 1658484765123456) .atNow(); - expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t\n", - ); - await sender.close(); - }); - - it("supports timestamp field as ns BigInt", async function () { - const sender = new Sender({ - protocol: "tcp", - protocol_version: "1", - host: "host", - init_buf_size: 1024, - }); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000000n, "ns") + .timestampColumn("ts", 1658484765123456, "us") + .atNow(); + await sender + .table("tableName") + .timestampColumn("ts", 1658484765123, "ms") .atNow(); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t\n", + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123000t\n", ); await sender.close(); }); - it("supports timestamp field as us BigInt", async function () { + it("supports timestamp field as BigInt with protocol v1", async function () { const sender = new Sender({ protocol: "tcp", protocol_version: "1", @@ -559,29 +526,57 @@ describe("Sender message builder test suite (anything not covered in client inte }); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000n, "us") + .timestampColumn("ts", 1658484765123456n) + .atNow(); + await sender + .table("tableName") + .timestampColumn("ts", 1658484765123456789n, "ns") + .atNow(); + await sender + .table("tableName") + .timestampColumn("ts", 1658484765123456n, "us") + .atNow(); + await sender + .table("tableName") + .timestampColumn("ts", 1658484765123n, "ms") .atNow(); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t\n", + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123000t\n", ); await sender.close(); }); - it("supports timestamp field as ms BigInt", async function () { + it("supports timestamp field as BigInt with protocol v2", async function () { const sender = new Sender({ protocol: "tcp", - protocol_version: "1", + protocol_version: "2", host: "host", init_buf_size: 1024, }); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000n, "ms") + .timestampColumn("ts", 1658484765123456n) + .atNow(); + await sender + .table("tableName") + .timestampColumn("ts", 1658484765123456789n, "ns") + .atNow(); + await sender + .table("tableName") + .timestampColumn("ts", 1658484765123456n, "us") + .atNow(); + await sender + .table("tableName") + .timestampColumn("ts", 1658484765123n, "ms") .atNow(); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t\n", + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123456789n\n" + + "tableName ts=1658484765123456t\n" + + "tableName ts=1658484765123000t\n", ); await sender.close(); }); @@ -593,123 +588,137 @@ describe("Sender message builder test suite (anything not covered in client inte host: "host", init_buf_size: 1024, }); - try { - await sender - .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) - // @ts-expect-error - Testing invalid options - .at(1658484769000000, "foobar"); - } catch (err) { - expect(err.message).toBe("Unknown timestamp unit: foobar"); - } + + await expect( + async () => + await sender + .table("tableName") + .booleanColumn("boolCol", true) + .timestampColumn("timestampCol", 1658484765000000) + // @ts-expect-error - Testing invalid timestamp unit + .at(1658484769000000, "foobar"), + ).rejects.toThrow("Unknown timestamp unit: foobar"); await sender.close(); }); - it("supports setting designated us timestamp as number from client", async function () { + it("supports designated timestamp as number for 'us' and 'ms' units with protocol v1", async function () { const sender = new Sender({ protocol: "tcp", protocol_version: "1", host: "host", init_buf_size: 1024, }); + await expect( + async () => + await sender + .table("tableName") + .intColumn("c1", 42) + .at(1658484769000000, "ns"), + ).rejects.toThrow( + "Designated timestamp must be a BigInt if it is set in nanoseconds", + ); + + sender.reset(); + await sender.table("tableName").intColumn("c1", 42).at(1658484769000000); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) + .intColumn("c1", 42) .at(1658484769000000, "us"); + await sender.table("tableName").intColumn("c1", 42).at(1658484769000, "ms"); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t 1658484769000000000\n", + "tableName c1=42i 1658484769000000000\n" + + "tableName c1=42i 1658484769000000000\n" + + "tableName c1=42i 1658484769000000000\n", ); await sender.close(); }); - it("supports setting designated ms timestamp as number from client", async function () { + it("supports designated timestamp as number for 'us' and 'ms' units with protocol v2", async function () { const sender = new Sender({ protocol: "tcp", - protocol_version: "1", + protocol_version: "2", host: "host", init_buf_size: 1024, }); - await sender - .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) - .at(1658484769000, "ms"); - expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t 1658484769000000000\n", + await expect( + async () => + await sender + .table("tableName") + .intColumn("c1", 42) + .at(1658484769000000, "ns"), + ).rejects.toThrow( + "Designated timestamp must be a BigInt if it is set in nanoseconds", ); - await sender.close(); - }); - it("supports setting designated timestamp as BigInt from client", async function () { - const sender = new Sender({ - protocol: "tcp", - protocol_version: "1", - host: "host", - init_buf_size: 1024, - }); + sender.reset(); + await sender.table("tableName").intColumn("c1", 42).at(1658484769000000); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) - .at(1658484769000000n); + .intColumn("c1", 42) + .at(1658484769000000, "us"); + await sender.table("tableName").intColumn("c1", 42).at(1658484769000, "ms"); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t 1658484769000000000\n", + "tableName c1=42i 1658484769000000t\n" + + "tableName c1=42i 1658484769000000t\n" + + "tableName c1=42i 1658484769000000t\n", ); await sender.close(); }); - it("supports setting designated ns timestamp as BigInt from client", async function () { + it("supports designated timestamp as BigInt with protocol v1", async function () { const sender = new Sender({ protocol: "tcp", protocol_version: "1", host: "host", init_buf_size: 1024, }); + await sender.table("tableName").intColumn("c1", 42).at(1658484769000000n); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) - .at(1658484769000000123n, "ns"); - expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t 1658484769000000123\n", - ); - await sender.close(); - }); - - it("supports setting designated us timestamp as BigInt from client", async function () { - const sender = new Sender({ - protocol: "tcp", - protocol_version: "1", - host: "host", - init_buf_size: 1024, - }); + .intColumn("c1", 42) + .at(1658484769000000n, "ns"); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) + .intColumn("c1", 42) .at(1658484769000000n, "us"); + await sender + .table("tableName") + .intColumn("c1", 42) + .at(1658484769000n, "ms"); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t 1658484769000000000\n", + "tableName c1=42i 1658484769000000000\n" + + "tableName c1=42i 1658484769000000\n" + + "tableName c1=42i 1658484769000000000\n" + + "tableName c1=42i 1658484769000000000\n", ); await sender.close(); }); - it("supports setting designated ms timestamp as BigInt from client", async function () { + it("supports designated timestamp as BigInt with protocol v2", async function () { const sender = new Sender({ protocol: "tcp", - protocol_version: "1", + protocol_version: "2", host: "host", init_buf_size: 1024, }); + await sender.table("tableName").intColumn("c1", 42).at(1658484769000000n); await sender .table("tableName") - .booleanColumn("boolCol", true) - .timestampColumn("timestampCol", 1658484765000000) + .intColumn("c1", 42) + .at(1658484769000000n, "ns"); + await sender + .table("tableName") + .intColumn("c1", 42) + .at(1658484769000000n, "us"); + await sender + .table("tableName") + .intColumn("c1", 42) .at(1658484769000n, "ms"); expect(bufferContent(sender)).toBe( - "tableName boolCol=t,timestampCol=1658484765000000t 1658484769000000000\n", + "tableName c1=42i 1658484769000000t\n" + + "tableName c1=42i 1658484769000000n\n" + + "tableName c1=42i 1658484769000000t\n" + + "tableName c1=42i 1658484769000000t\n", ); await sender.close(); }); @@ -948,7 +957,7 @@ describe("Sender message builder test suite (anything not covered in client inte }); expect(() => sender.table("tableName").timestampColumn("intField", 123.222), - ).toThrow("Value must be an integer or BigInt, received 123.222"); + ).toThrow("Timestamp value must be an integer or BigInt, received 123.222"); await sender.close(); }); diff --git a/test/sender.integration.test.ts b/test/sender.integration.test.ts index 397c6c3..ee7a2fd 100644 --- a/test/sender.integration.test.ts +++ b/test/sender.integration.test.ts @@ -3,8 +3,7 @@ import { describe, it, expect, beforeAll, afterAll } from "vitest"; import { GenericContainer, StartedTestContainer } from "testcontainers"; import http from "http"; -import { Sender } from "../src"; -import { SenderOptions } from "../src/options"; +import { Sender, SenderOptions } from "../src"; const HTTP_OK = 200; diff --git a/test/sender.transport.test.ts b/test/sender.transport.test.ts index 6bf27ea..56759d1 100644 --- a/test/sender.transport.test.ts +++ b/test/sender.transport.test.ts @@ -4,10 +4,7 @@ import { readFileSync } from "fs"; import { Agent } from "undici"; import http from "http"; -import { Sender } from "../src"; -import { SenderOptions } from "../src/options"; -import { UndiciTransport } from "../src/transport/http/undici"; -import { HttpTransport } from "../src/transport/http/stdlib"; +import { Sender, SenderOptions, UndiciTransport, HttpTransport } from "../src"; import { MockProxy } from "./util/mockproxy"; import { MockHttp } from "./util/mockhttp"; diff --git a/test/testapp.ts b/test/testapp.ts index 39f1267..aed35eb 100644 --- a/test/testapp.ts +++ b/test/testapp.ts @@ -1,8 +1,7 @@ import { readFileSync } from "node:fs"; import { Proxy } from "./util/proxy"; -import { Sender } from "../src"; -import { SenderOptions } from "../src/options"; +import { Sender, SenderOptions } from "../src"; const PROXY_PORT = 9099; const PORT = 9009;