Skip to content

feat(nodejs): array support #50

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: binary_protocol_arrays
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .github/workflows/build.yml
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@ on:
branches:
- main
pull_request:
schedule:
- cron: '15 2,10,18 * * *'

jobs:
test:
Expand Down
89 changes: 72 additions & 17 deletions src/buffer/base.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import {
DEFAULT_MAX_BUFFER_SIZE,
} from "./index";
import {
ArrayPrimitive,
isInteger,
timestampToMicros,
timestampToNanos,
Expand Down Expand Up @@ -232,6 +233,8 @@ abstract class SenderBufferBase implements SenderBuffer {
*/
abstract floatColumn(name: string, value: number): SenderBuffer;

abstract arrayColumn(name: string, value: unknown[]): SenderBuffer;

/**
* Write an integer column with its value into the buffer.
*
Expand Down Expand Up @@ -373,36 +376,60 @@ abstract class SenderBufferBase implements SenderBuffer {
this.writeEscaped(name);
this.write("=");
writeValue();
this.assertBufferOverflow();
this.hasColumns = true;
}

protected write(data: string) {
this.position += this.buffer.write(data, this.position);
if (this.position > this.bufferSize) {
// should never happen, if checkCapacity() is correctly used
throw new Error(
`Buffer overflow [position=${this.position}, bufferSize=${this.bufferSize}]`,
);
}
}

protected writeByte(data: number) {
this.position = this.buffer.writeInt8(data, this.position);
if (this.position > this.bufferSize) {
// should never happen, if checkCapacity() is correctly used
throw new Error(
`Buffer overflow [position=${this.position}, bufferSize=${this.bufferSize}]`,
);
}
}

protected writeInt(data: number) {
this.position = this.buffer.writeInt32LE(data, this.position);
}

protected writeDouble(data: number) {
this.position = this.buffer.writeDoubleLE(data, this.position);
if (this.position > this.bufferSize) {
// should never happen, if checkCapacity() is correctly used
throw new Error(
`Buffer overflow [position=${this.position}, bufferSize=${this.bufferSize}]`,
);
}

protected writeArray(
arr: unknown[],
dimensions: number[],
type: ArrayPrimitive,
) {
this.checkCapacity([], 1 + dimensions.length * 4);
this.writeByte(dimensions.length);
for (let i = 0; i < dimensions.length; i++) {
this.writeInt(dimensions[i]);
}

this.checkCapacity([], SenderBufferBase.arraySize(dimensions, type));
this.writeArrayValues(arr, dimensions);
}

private writeArrayValues(arr: unknown[], dimensions: number[]) {
if (Array.isArray(arr[0])) {
for (let i = 0; i < arr.length; i++) {
this.writeArrayValues(arr[i] as unknown[], dimensions);
}
} else {
const type = typeof arr[0];
switch (type) {
case "number":
for (let i = 0; i < arr.length; i++) {
this.position = this.buffer.writeDoubleLE(
arr[i] as number,
this.position,
);
}
break;
default:
throw new Error(`Unsupported array type [type=${type}]`);
}
}
}

Expand Down Expand Up @@ -442,6 +469,34 @@ abstract class SenderBufferBase implements SenderBuffer {
}
}
}

private static arraySize(dimensions: number[], type: ArrayPrimitive): number {
let numOfElements = 1;
for (let i = 0; i < dimensions.length; i++) {
numOfElements *= dimensions[i];
}

switch (type) {
case "number":
return numOfElements * 8;
case "boolean":
return numOfElements;
case "string":
// in case of string[] capacity check is done separately for each array element
return 0;
default:
throw new Error(`Unsupported array type [type=${type}]`);
}
}

private assertBufferOverflow() {
if (this.position > this.bufferSize) {
// should never happen, if checkCapacity() is correctly used
throw new Error(
`Buffer overflow [position=${this.position}, bufferSize=${this.bufferSize}]`,
);
}
}
}

export { SenderBufferBase };
4 changes: 4 additions & 0 deletions src/buffer/bufferv1.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,10 @@ class SenderBufferV1 extends SenderBufferBase {
);
return this;
}

arrayColumn(): SenderBuffer {
throw new Error("Arrays are not supported in protocol v1");
}
}

export { SenderBufferV1 };
29 changes: 29 additions & 0 deletions src/buffer/bufferv2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,14 @@
import { SenderOptions } from "../options";
import { SenderBuffer } from "./index";
import { SenderBufferBase } from "./base";
import { getDimensions, validateArray } from "../utils";

const COLUMN_TYPE_DOUBLE: number = 10;
const COLUMN_TYPE_NULL: number = 33;

const ENTITY_TYPE_ARRAY: number = 14;
const ENTITY_TYPE_DOUBLE: number = 16;

const EQUALS_SIGN: number = "=".charCodeAt(0);

/**
Expand Down Expand Up @@ -36,6 +42,29 @@ class SenderBufferV2 extends SenderBufferBase {
);
return this;
}

arrayColumn(name: string, value: unknown[]): SenderBuffer {
const dimensions = getDimensions(value);
const type = validateArray(value, dimensions);
// only number arrays and NULL supported for now
if (type !== "number" && type !== null) {
throw new Error(`Unsupported array type [type=${type}]`);
}

this.writeColumn(name, value, () => {
this.checkCapacity([], 3);
this.writeByte(EQUALS_SIGN);
this.writeByte(ENTITY_TYPE_ARRAY);

if (!value) {
this.writeByte(COLUMN_TYPE_NULL);
} else {
this.writeByte(COLUMN_TYPE_DOUBLE);
this.writeArray(value, dimensions, type);
}
});
return this;
}
}

export { SenderBufferV2 };
2 changes: 2 additions & 0 deletions src/buffer/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,8 @@ interface SenderBuffer {
*/
floatColumn(name: string, value: number): SenderBuffer;

arrayColumn(name: string, value: unknown[]): SenderBuffer;

/**
* Write an integer column with its value into the buffer.
*
Expand Down
5 changes: 5 additions & 0 deletions src/sender.ts
Original file line number Diff line number Diff line change
Expand Up @@ -241,6 +241,11 @@ class Sender {
return this;
}

arrayColumn(name: string, value: unknown[]): Sender {
this.buffer.arrayColumn(name, value);
return this;
}

/**
* Write an integer column with its value into the buffer of the sender.
*
Expand Down
72 changes: 72 additions & 0 deletions src/utils.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
type ArrayPrimitive = "number" | "boolean" | "string" | null;

type TimestampUnit = "ns" | "us" | "ms";

function isBoolean(value: unknown): value is boolean {
Expand Down Expand Up @@ -36,6 +38,73 @@ function timestampToNanos(timestamp: bigint, unit: TimestampUnit) {
}
}

function getDimensions(data: unknown) {
const dimensions: number[] = [];
while (Array.isArray(data)) {
if (data.length === 0) {
throw new Error("Zero length array not supported");
}
dimensions.push(data.length);
data = data[0];
}
return dimensions;
}

function validateArray(data: unknown[], dimensions: number[]): ArrayPrimitive {
if (data === null || data === undefined) {
return null;
}
if (!Array.isArray(data)) {
throw new Error(
`The value must be an array [value=${JSON.stringify(data)}, type=${typeof data}]`,
);
}

let expectedType: ArrayPrimitive = null;

function checkArray(
array: unknown[],
depth: number = 0,
path: string = "",
): void {
const expectedLength = dimensions[depth];
if (array.length !== expectedLength) {
throw new Error(
`Length of arrays do not match [expected=${expectedLength}, actual=${array.length}, dimensions=[${dimensions}], path=${path}]`,
);
}

if (depth < dimensions.length - 1) {
// intermediate level, expecting arrays
for (let i = 0; i < array.length; i++) {
if (!Array.isArray(array[i])) {
throw new Error(
`Mixed types found [expected=array, current=${typeof array[i]}, path=${path}[${i}]]`,
);
}
checkArray(array[i] as unknown[], depth + 1, `${path}[${i}]`);
}
} else {
// leaf level, expecting primitives
if (expectedType === null) {
expectedType = typeof array[0] as ArrayPrimitive;
}

for (let i = 0; i < array.length; i++) {
const currentType = typeof array[i] as ArrayPrimitive;
if (currentType !== expectedType) {
throw new Error(
`Mixed types found [expected=${expectedType}, current=${currentType}, path=${path}[${i}]]`,
);
}
}
}
}

checkArray(data);
return expectedType;
}

/**
* Fetches JSON data from a URL.
* @template T - The expected type of the JSON response
Expand Down Expand Up @@ -66,4 +135,7 @@ export {
timestampToNanos,
TimestampUnit,
fetchJson,
getDimensions,
validateArray,
ArrayPrimitive,
};
Loading