feat: support for DECIMAL type

README.md

@@ -312,6 +312,60 @@ function rndInt(limit: number) {
 run().then(console.log).catch(console.error);
 ```
 
+### Decimal usage example
+
+Since v9.2.0, QuestDB supports the DECIMAL data type.
+Decimals can be ingested with ILP protocol v3 using either textual or binary representation.
+
+#### Textual representation
+
+```typescript
+import { Sender } from "@questdb/nodejs-client";
+
+async function runDecimals() {
+  const sender = await Sender.fromConfig(
+    "tcp::addr=127.0.0.1:9009;protocol_version=3",
+  );
+
+  await sender
+    .table("fx")
+    // textual ILP form keeps the literal and its exact scale
+    .decimalColumnText("mid", "1.234500")
+    .atNow();
+
+  await sender.flush();
+  await sender.close();
+}
+
+runDecimals().catch(console.error);
+// Resulting ILP line: fx mid=1.234500d
+```
+
+#### Binary representation
+
+It is recommended to use the binary representation for better ingestion performance and reduced payload size (for bigger decimals).
+
+```typescript
+import { Sender } from "@questdb/nodejs-client";
+
+async function runDecimals() {
+  const sender = await Sender.fromConfig(
+    "tcp::addr=127.0.0.1:9009;protocol_version=3",
+  );
+
+  await sender
+    .table("fx")
+    // textual ILP form keeps the literal and its exact scale
+    .decimalColumn("mid", 123456n, 3) // 123456 * 10^-3 = 123.456
+    .atNow();
+
+  await sender.flush();
+  await sender.close();
+}
+
+runDecimals().catch(console.error);
+```
+
 ## Community
 
 If you need help, have additional questions or want to provide feedback, you

src/buffer/base.ts

@@ -522,6 +522,43 @@ abstract class SenderBufferBase implements SenderBuffer {
       }
     }
   }
+
+  /* eslint-disable @typescript-eslint/no-unused-vars */
+  /**
+   * Writes a decimal value into the buffer using the text format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} value - Column value, accepts only number/string values.
+   * @returns {Sender} Returns with a reference to this buffer.
+   * @throws Error if decimals are not supported by the buffer implementation, or decimal validation fails:
+   * - string value is not a valid decimal representation
+   */
+  decimalColumnText(name: string, value: string | number): SenderBuffer {
+    throw new Error("Decimals are not supported in protocol v1/v2");
+  }
+
+  /**
+   * Writes a decimal value into the buffer using the binary format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} unscaled - The unscaled value of the decimal in two's
+   * complement representation and big-endian byte order.
+   * An empty array represents the NULL value.
+   * @param {number} scale - The scale of the decimal value.
+   * @returns {Sender} Returns with a reference to this buffer.
+   */
+  decimalColumn(
+    name: string,
+    unscaled: Int8Array | bigint,
+    scale: number,
+  ): SenderBuffer {
+    throw new Error("Decimals are not supported in protocol v1/v2");
+  }
+  /* eslint-enable @typescript-eslint/no-unused-vars */
 }
 
 export { SenderBufferBase };

src/buffer/bufferv3.ts

@@ -0,0 +1,122 @@
+// @ts-check
+import { SenderOptions } from "../options";
+import { SenderBuffer } from "./index";
+import { bigintToTwosComplementBytes } from "../utils";
+import { SenderBufferV2 } from "./bufferv2";
+import { validateDecimalText } from "../validation";
+
+// Entity type constants for protocol v3.
+const ENTITY_TYPE_DECIMAL: number = 23;
+
+// ASCII code for equals sign used in binary protocol.
+const EQUALS_SIGN: number = "=".charCodeAt(0);
+
+/**
+ * Buffer implementation for protocol version 3.
+ *
+ * Provides support for decimals.
+ */
+class SenderBufferV3 extends SenderBufferV2 {
+  /**
+   * Creates a new SenderBufferV3 instance.
+   *
+   * @param {SenderOptions} options - Sender configuration object.
+   *
+   * See SenderOptions documentation for detailed description of configuration options.
+   */
+  constructor(options: SenderOptions) {
+    super(options);
+  }
+
+  /**
+   * Writes a decimal value into the buffer using the text format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} value - Column value, accepts only number/string values.
+   * @returns {Sender} Returns with a reference to this buffer.
+   * @throws Error if decimals are not supported by the buffer implementation, or decimal validation fails:
+   * - string value is not a valid decimal representation
+   */
+  decimalColumnText(name: string, value: string | number): SenderBuffer {
+    let str = "";
+    if (typeof value === "string") {
+      validateDecimalText(value);
+      str = value;
+    } else if (typeof value === "number") {
+      str = value.toString();
+    } else {
+      throw new TypeError(`Invalid decimal value type: ${typeof value}`);
+    }
+    this.writeColumn(name, str, () => {
+      this.checkCapacity([str], 1);
+      this.write(str);
+      this.write("d");
+    });
+    return this;
+  }
+
+  /**
+   * Writes a decimal value into the buffer using the binary format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} unscaled - The unscaled value of the decimal in two's
+   * complement representation and big-endian byte order.
+   * An empty array represents the NULL value.
+   * @param {number} scale - The scale of the decimal value.
+   * @returns {Sender} Returns with a reference to this buffer.
+   * @throws Error if decimals are not supported by the buffer implementation, or decimal validation fails:
+   * - unscaled value length is not between 0 and 32 bytes
+   * - scale is not between 0 and 76
+   * - unscaled value contains invalid bytes
+   */
+  decimalColumn(
+    name: string,
+    unscaled: Int8Array | bigint,
+    scale: number,
+  ): SenderBuffer {
+    if (scale < 0 || scale > 76) {
+      throw new RangeError("Scale must be between 0 and 76");
+    }
+    let arr: number[];
+    if (typeof unscaled === "bigint") {
+      arr = bigintToTwosComplementBytes(unscaled);
+    } else if (unscaled instanceof Int8Array) {
+      arr = Array.from(unscaled);
+    } else {
+      throw new TypeError(
+        `Invalid unscaled value type: ${typeof unscaled}, expected Int8Array or bigint`,
+      );
+    }
+    if (arr.length > 32) {
+      throw new RangeError(
+        "Unscaled value length must be between 0 and 32 bytes",
+      );
+    }
+    this.writeColumn(name, unscaled, () => {
+      this.checkCapacity([], 4 + arr.length);
+      this.writeByte(EQUALS_SIGN);
+      this.writeByte(ENTITY_TYPE_DECIMAL);
+      this.writeByte(scale);
+      this.writeByte(arr.length);
+      for (let i = 0; i < arr.length; i++) {
+        let byte = arr[i];
+        if (byte > 255 || byte < -128) {
+          throw new RangeError(
+            `Unscaled value contains invalid byte [index=${i}, value=${byte}]`,
+          );
+        }
+        if (byte > 127) {
+          byte -= 256;
+        }
+        this.writeByte(byte);
+      }
+    });
+    return this;
+  }
+}
+
+export { SenderBufferV3 };

src/buffer/index.ts

@@ -6,10 +6,12 @@ import {
   PROTOCOL_VERSION_V1,
   PROTOCOL_VERSION_V2,
   PROTOCOL_VERSION_AUTO,
+  PROTOCOL_VERSION_V3,
 } from "../options";
 import { TimestampUnit } from "../utils";
 import { SenderBufferV1 } from "./bufferv1";
 import { SenderBufferV2 } from "./bufferv2";
+import { SenderBufferV3 } from "./bufferv3";
 
 // Default initial buffer size in bytes (64 KB).
 const DEFAULT_BUFFER_SIZE = 65536; //  64 KB
@@ -26,6 +28,8 @@ const DEFAULT_MAX_BUFFER_SIZE = 104857600; // 100 MB
  */
 function createBuffer(options: SenderOptions): SenderBuffer {
   switch (options.protocol_version) {
+    case PROTOCOL_VERSION_V3:
+      return new SenderBufferV3(options);
     case PROTOCOL_VERSION_V2:
       return new SenderBufferV2(options);
     case PROTOCOL_VERSION_V1:
@@ -170,6 +174,35 @@ interface SenderBuffer {
     unit: TimestampUnit,
   ): SenderBuffer;
 
+  /**
+   * Writes a decimal value into the buffer using the text format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} value - Column value, accepts only number/string values.
+   * @returns {Sender} Returns with a reference to this buffer.
+   */
+  decimalColumnText(name: string, value: string | number): SenderBuffer;
+
+  /**
+   * Writes a decimal value into the buffer using the binary format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} unscaled - The unscaled value of the decimal in two's
+   * complement representation and big-endian byte order.
+   * An empty array represents the NULL value.
+   * @param {number} scale - The scale of the decimal value.
+   * @returns {Sender} Returns with a reference to this buffer.
+   */
+  decimalColumn(
+    name: string,
+    unscaled: Int8Array | bigint,
+    scale: number,
+  ): SenderBuffer;
+
   /**
    * Closes the row after writing the designated timestamp into the buffer.
    *

src/index.ts

@@ -17,3 +17,4 @@ export { TcpTransport } from "./transport/tcp";
 export { HttpTransport } from "./transport/http/stdlib";
 export { UndiciTransport } from "./transport/http/undici";
 export type { Logger } from "./logging";
+export { bigintToTwosComplementBytes } from "./utils";

src/options.ts

@@ -23,6 +23,7 @@ const UNSAFE_OFF = "unsafe_off";
 const PROTOCOL_VERSION_AUTO = "auto";
 const PROTOCOL_VERSION_V1 = "1";
 const PROTOCOL_VERSION_V2 = "2";
+const PROTOCOL_VERSION_V3 = "3";
 
 const LINE_PROTO_SUPPORT_VERSION = "line.proto.support.versions";
 
@@ -258,6 +259,8 @@ class SenderOptions {
 
     if (supportedVersions.length === 0) {
       options.protocol_version = PROTOCOL_VERSION_V1;
+    } else if (supportedVersions.includes(PROTOCOL_VERSION_V3)) {
+      options.protocol_version = PROTOCOL_VERSION_V3;
     } else if (supportedVersions.includes(PROTOCOL_VERSION_V2)) {
       options.protocol_version = PROTOCOL_VERSION_V2;
     } else if (supportedVersions.includes(PROTOCOL_VERSION_V1)) {
@@ -488,10 +491,11 @@ function parseProtocolVersion(options: SenderOptions) {
       break;
     case PROTOCOL_VERSION_V1:
     case PROTOCOL_VERSION_V2:
+    case PROTOCOL_VERSION_V3:
       break;
     default:
       throw new Error(
-        `Invalid protocol version: '${protocol_version}', accepted values: 'auto', '1', '2'`,
+        `Invalid protocol version: '${protocol_version}', accepted values: 'auto', '1', '2', '3'`,
       );
   }
   return;
@@ -628,4 +632,5 @@ export {
   PROTOCOL_VERSION_AUTO,
   PROTOCOL_VERSION_V1,
   PROTOCOL_VERSION_V2,
+  PROTOCOL_VERSION_V3,
 };

src/sender.ts

@@ -342,6 +342,47 @@ class Sender {
     return this;
   }
 
+  /**
+   * Writes a decimal value into the buffer using the text format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} value - Column value, accepts only number/string values.
+   * @returns {Sender} Returns with a reference to this buffer.
+   * @throws Error if decimals are not supported by the buffer implementation, or decimal validation fails:
+   * - string value is not a valid decimal representation
+   */
+  decimalColumnText(name: string, value: string | number): Sender {
+    this.buffer.decimalColumnText(name, value);
+    return this;
+  }
+
+  /**
+   * Writes a decimal value into the buffer using the binary format.
+   *
+   * Use it to insert into DECIMAL database columns.
+   *
+   * @param {string} name - Column name.
+   * @param {number} unscaled - The unscaled value of the decimal in two's
+   * complement representation and big-endian byte order.
+   * An empty array represents the NULL value.
+   * @param {number} scale - The scale of the decimal value.
+   * @returns {Sender} Returns with a reference to this buffer.
+   * @throws Error if decimals are not supported by the buffer implementation, or decimal validation fails:
+   * - unscaled value length is not between 0 and 32 bytes
+   * - scale is not between 0 and 76
+   * - unscaled value contains invalid bytes
+   */
+  decimalColumn(
+    name: string,
+    unscaled: Int8Array | bigint,
+    scale: number,
+  ): Sender {
+    this.buffer.decimalColumn(name, unscaled, scale);
+    return this;
+  }
+
   /**
    * Closes the row after writing the designated timestamp into the buffer of the sender.
    *