This is an automated email from the ASF dual-hosted git repository. CritasWang pushed a commit to branch perf/serialization-single-pass in repository https://gitbox.apache.org/repos/asf/iotdb-client-nodejs.git
commit 263a4e48f0e71eb2cbb3c57c6308de6a5f448440 Author: CritasWang <[email protected]> AuthorDate: Tue Jul 14 17:53:48 2026 +0800 perf: single-pass single-buffer tablet serialization, BigInt-free int64 writes CPU profiling of the write path (insertTablet-heavy workload) showed the event loop 79.5% busy with serializeTabletValues at 26% self time and GC at 24.5% — almost entirely allocation churn in the serializer. This rewrites the fast (default) write path: - serializeTabletValues: new single-pass serializeTabletValuesFast() writes all columns AND null bitmaps into ONE exactly-sized buffer. Eliminates the rows->columns transpose (one array per column), the per-column boolean[] null bitmaps (built then discarded when a column has no nulls), the per-column intermediate buffers, the per-column 1-byte Buffer.from([flag]) allocations, and the trailing Buffer.concat (which re-copied the whole payload). Fixed-width sizes come from the schema; variable-width columns get one byteLength pre-scan. - INT64/TIMESTAMP writes: writeBigInt64BE(BigInt(v)) replaced with a sign-correct hi/lo 32-bit pair (hi = floor(v / 2^32), lo = v >>> 0), avoiding one BigInt allocation per value. BigInt inputs and non-safe integers still take the exact writeBigInt64BE path. - TEXT/STRING columns: single-entry encode cache — TAG columns repeat one string per tablet, so consecutive identical values reuse the encoded Buffer instead of re-encoding UTF-8 every row. - BufferPool: removed from the write path. It was acquire-only (zero release() calls anywhere in src/), i.e. a guaranteed 0% hit rate and pure overhead over Buffer.allocUnsafe. The class stays exported (public API) but is deprecated. Wire format is unchanged — existing serialization unit tests pass untouched, and new golden tests assert the fast path output equals the legacy path byte-for-byte for mixed-type tablets with nulls, all-null columns, bitmap-width edge cases (17 rows), BLOB/DATE, and negative / MIN_SAFE_INTEGER / BigInt int64 values. The legacy path (enableFastSerialization=false) is untouched. --- src/client/Session.ts | 22 +- src/index.ts | 6 +- src/utils/BufferPool.ts | 8 +- src/utils/FastSerializer.ts | 435 ++++++++++++++++++++++++++------- tests/unit/FastSerializer.test.ts | 127 +++++++++- tests/unit/TabletSerialization.test.ts | 85 +++++++ 6 files changed, 572 insertions(+), 111 deletions(-) diff --git a/src/client/Session.ts b/src/client/Session.ts index 879f62e..d16b30a 100644 --- a/src/client/Session.ts +++ b/src/client/Session.ts @@ -30,11 +30,10 @@ import { SessionDataSet } from "./SessionDataSet"; import { RowRecord } from "./RowRecord"; import { BaseColumnDecoder, ColumnEncoding, Column } from "./ColumnDecoder"; import { RedirectException } from "../utils/Errors"; -import { - serializeColumnFast, - serializeTimestamps +import { + serializeTabletValuesFast, + serializeTimestamps } from "../utils/FastSerializer"; -import { globalBufferPool } from "../utils/BufferPool"; const ttypes = require("../thrift/generated/client_types"); @@ -609,7 +608,15 @@ export class Session { dataTypes: number[], rowCount: number, ): Buffer { - // Serialize tablet values based on data types + // Fast path (default): single-pass, single-buffer serialization of all + // columns + null bitmaps — no transpose, no per-column intermediate + // buffers, no trailing Buffer.concat. Wire format is identical to the + // legacy path below. + if (this.config.enableFastSerialization) { + return serializeTabletValuesFast(values, dataTypes, rowCount); + } + + // Legacy path: per-column serialization + Buffer.concat // Format: all columns data, then bitmap for null values const buffers: Buffer[] = []; const bitMaps: (boolean[] | null)[] = []; @@ -633,10 +640,7 @@ export class Session { } } - // Use fast serialization if enabled, otherwise fall back to legacy - const buffer = this.config.enableFastSerialization - ? serializeColumnFast(columnValues, dataType) - : this.serializeColumn(columnValues, dataType); + const buffer = this.serializeColumn(columnValues, dataType); buffers.push(buffer); bitMaps.push(hasNull ? nullBitmap : null); } diff --git a/src/index.ts b/src/index.ts index 54fdee4..b27beaa 100644 --- a/src/index.ts +++ b/src/index.ts @@ -44,7 +44,11 @@ export { RedirectException, TSStatusCode } from "./utils/Errors"; export { RedirectCache } from "./client/RedirectCache"; export { enableGlobalCleanup } from "./utils/ProcessCleanup"; export { BufferPool, globalBufferPool } from "./utils/BufferPool"; -export { serializeColumnFast, serializeTimestamps } from "./utils/FastSerializer"; +export { + serializeColumnFast, + serializeTimestamps, + serializeTabletValuesFast, +} from "./utils/FastSerializer"; // Concurrent execution utilities for Node.js optimization export { diff --git a/src/utils/BufferPool.ts b/src/utils/BufferPool.ts index 11e32e0..f477f2d 100644 --- a/src/utils/BufferPool.ts +++ b/src/utils/BufferPool.ts @@ -22,7 +22,13 @@ import { logger } from "./Logger"; /** * Buffer pool for reusing buffers to reduce GC pressure * Inspired by pg nodejs client's buffer management strategy - * + * + * @deprecated No longer used internally. The write path never called + * release(), so the pool had a 0% hit rate and acquire() was pure overhead + * over Buffer.allocUnsafe. Kept only because it is part of the public API + * surface (exported from src/index.ts); may be removed in a future major + * version. + * * Key design principles: * 1. Size classes to minimize waste (powers of 2) * 2. Maximum pool size to prevent memory bloat diff --git a/src/utils/FastSerializer.ts b/src/utils/FastSerializer.ts index a87f9b3..d342be2 100644 --- a/src/utils/FastSerializer.ts +++ b/src/utils/FastSerializer.ts @@ -17,25 +17,40 @@ * under the License. */ -import { globalBufferPool } from "./BufferPool"; - /** * Fast serialization utilities for IoTDB data types * Optimized for performance with: - * - Pre-allocated buffers - * - Buffer pooling - * - Single-pass serialization + * - Single exact-size buffer allocation per tablet (no Buffer.concat re-copy) + * - BigInt-free int64 writes (hi/lo 32-bit pair) + * - Single-entry string encode cache (TAG columns repeat the same value) * - Minimal object allocation - * - * Inspired by pg nodejs client's serialization strategy */ +/** + * Write a 64-bit big-endian signed integer without allocating a BigInt. + * + * For any safe integer v (positive or negative): + * hi = Math.floor(v / 2^32) — arithmetic shift, sign-correct + * lo = v >>> 0 — ToUint32 performs the mod-2^32 reduction + * so that hi * 2^32 + lo === v, which is exactly the two's complement + * representation split into two 32-bit halves. + * + * Falls back to writeBigInt64BE for BigInt inputs and non-safe-integer + * numbers (preserves legacy error behavior for invalid values like 1.5). + */ +export function writeInt64BE(buffer: Buffer, v: number | bigint, offset: number): void { + if (typeof v === "number" && Number.isSafeInteger(v)) { + buffer.writeInt32BE(Math.floor(v / 0x100000000), offset); + buffer.writeUInt32BE(v >>> 0, offset + 4); + } else { + buffer.writeBigInt64BE(typeof v === "bigint" ? v : BigInt(v as any), offset); + } +} + /** * Serialize BOOLEAN column (1 byte per value) - * Optimized: Single buffer allocation */ export function serializeBooleanColumn(values: any[]): Buffer { - // For small buffers, direct allocation is faster than pooling const buffer = Buffer.allocUnsafe(values.length); for (let i = 0; i < values.length; i++) { const v = values[i]; @@ -46,134 +61,133 @@ export function serializeBooleanColumn(values: any[]): Buffer { /** * Serialize INT32 column (4 bytes per value, big-endian) - * Optimized: Single buffer allocation with direct writes */ export function serializeInt32Column(values: any[]): Buffer { - const size = values.length * 4; - const buffer = size >= 1024 ? globalBufferPool.acquire(size) : Buffer.allocUnsafe(size); - + const buffer = Buffer.allocUnsafe(values.length * 4); for (let i = 0; i < values.length; i++) { const v = values[i]; buffer.writeInt32BE(v === null || v === undefined ? 0 : v, i * 4); } - - return buffer.subarray(0, size); + return buffer; } /** * Serialize INT64 column (8 bytes per value, big-endian) - * Optimized: Single buffer allocation with direct writes + * Optimized: BigInt-free hi/lo writes for number inputs */ export function serializeInt64Column(values: any[]): Buffer { - const size = values.length * 8; - const buffer = size >= 1024 ? globalBufferPool.acquire(size) : Buffer.allocUnsafe(size); - + const buffer = Buffer.allocUnsafe(values.length * 8); for (let i = 0; i < values.length; i++) { const v = values[i]; - buffer.writeBigInt64BE( - v === null || v === undefined ? BigInt(0) : BigInt(v), - i * 8 - ); + if (v === null || v === undefined) { + buffer.writeInt32BE(0, i * 8); + buffer.writeUInt32BE(0, i * 8 + 4); + } else { + writeInt64BE(buffer, v, i * 8); + } } - - return buffer.subarray(0, size); + return buffer; } /** * Serialize FLOAT column (4 bytes per value, big-endian) - * Optimized: Single buffer allocation with direct writes */ export function serializeFloatColumn(values: any[]): Buffer { - const size = values.length * 4; - const buffer = size >= 1024 ? globalBufferPool.acquire(size) : Buffer.allocUnsafe(size); - + const buffer = Buffer.allocUnsafe(values.length * 4); for (let i = 0; i < values.length; i++) { const v = values[i]; buffer.writeFloatBE(v === null || v === undefined ? 0.0 : v, i * 4); } - - return buffer.subarray(0, size); + return buffer; } /** * Serialize DOUBLE column (8 bytes per value, big-endian) - * Optimized: Single buffer allocation with direct writes */ export function serializeDoubleColumn(values: any[]): Buffer { - const size = values.length * 8; - const buffer = size >= 1024 ? globalBufferPool.acquire(size) : Buffer.allocUnsafe(size); - + const buffer = Buffer.allocUnsafe(values.length * 8); for (let i = 0; i < values.length; i++) { const v = values[i]; buffer.writeDoubleBE(v === null || v === undefined ? 0.0 : v, i * 8); } - - return buffer.subarray(0, size); + return buffer; } /** * Serialize TEXT/STRING column (4 bytes length + UTF-8 bytes per value) - * Optimized: Two-pass approach to pre-calculate total size + * Optimized: single-entry encode cache — TAG columns repeat the same string + * for every row of a tablet, so the encoded Buffer is reused on consecutive + * identical values instead of re-encoding. */ export function serializeTextColumn(values: any[]): Buffer { - // Phase 1: Calculate total size and prepare string buffers - const strBuffers: Buffer[] = []; + // Phase 1: Calculate total size (byteLength only — no Buffer allocation) let totalSize = 0; - + let lastSizeStr: string | null = null; + let lastSizeLen = 0; + for (const v of values) { - const str = v === null || v === undefined ? "" : String(v); - const strBytes = Buffer.from(str, "utf8"); - strBuffers.push(strBytes); - totalSize += 4 + strBytes.length; + if (v === null || v === undefined) { + totalSize += 4; + continue; + } + const str = typeof v === "string" ? v : String(v); + if (str !== lastSizeStr) { + lastSizeLen = Buffer.byteLength(str, "utf8"); + lastSizeStr = str; + } + totalSize += 4 + lastSizeLen; } - - // Phase 2: Single allocation and copy - const result = totalSize >= 1024 ? globalBufferPool.acquire(totalSize) : Buffer.allocUnsafe(totalSize); + + // Phase 2: Single allocation, encode with single-entry cache + const result = Buffer.allocUnsafe(totalSize); let offset = 0; - - for (const strBytes of strBuffers) { - result.writeInt32BE(strBytes.length, offset); + let lastStr: string | null = null; + let lastBuf: Buffer | null = null; + + for (const v of values) { + if (v === null || v === undefined) { + result.writeInt32BE(0, offset); + offset += 4; + continue; + } + const str = typeof v === "string" ? v : String(v); + if (str !== lastStr || lastBuf === null) { + lastBuf = Buffer.from(str, "utf8"); + lastStr = str; + } + result.writeInt32BE(lastBuf.length, offset); offset += 4; - strBytes.copy(result, offset); - offset += strBytes.length; + lastBuf.copy(result, offset); + offset += lastBuf.length; } - - return result.subarray(0, totalSize); + + return result; } /** * Serialize TIMESTAMP column (8 bytes per value, big-endian) * Handles both Date objects and numeric timestamps - * Optimized: Single buffer allocation with direct writes + * Optimized: BigInt-free hi/lo writes */ export function serializeTimestampColumn(values: any[]): Buffer { - const size = values.length * 8; - const buffer = size >= 1024 ? globalBufferPool.acquire(size) : Buffer.allocUnsafe(size); - + const buffer = Buffer.allocUnsafe(values.length * 8); for (let i = 0; i < values.length; i++) { const v = values[i]; - let timestamp = BigInt(0); - if (v !== null && v !== undefined) { - if (v instanceof Date) { - timestamp = BigInt(v.getTime()); - } else { - timestamp = BigInt(v); - } + if (v === null || v === undefined) { + buffer.writeInt32BE(0, i * 8); + buffer.writeUInt32BE(0, i * 8 + 4); + } else { + writeInt64BE(buffer, v instanceof Date ? v.getTime() : v, i * 8); } - buffer.writeBigInt64BE(timestamp, i * 8); } - - return buffer.subarray(0, size); + return buffer; } /** * Serialize DATE column (4 bytes per value, days since epoch) - * Optimized: Single buffer allocation with direct writes */ export function serializeDateColumn(values: any[]): Buffer { - const size = values.length * 4; - const buffer = size >= 1024 ? globalBufferPool.acquire(size) : Buffer.allocUnsafe(size); - + const buffer = Buffer.allocUnsafe(values.length * 4); for (let i = 0; i < values.length; i++) { const v = values[i]; let days = 0; @@ -186,19 +200,18 @@ export function serializeDateColumn(values: any[]): Buffer { } buffer.writeInt32BE(days, i * 4); } - - return buffer.subarray(0, size); + return buffer; } /** * Serialize BLOB column (4 bytes length + binary data per value) - * Optimized: Two-pass approach to pre-calculate total size + * Two-pass approach to pre-calculate total size */ export function serializeBlobColumn(values: any[]): Buffer { // Phase 1: Calculate total size and prepare blob buffers const blobBuffers: Buffer[] = []; let totalSize = 0; - + for (const v of values) { const blob = v === null || v === undefined ? Buffer.alloc(0) @@ -208,38 +221,37 @@ export function serializeBlobColumn(values: any[]): Buffer { blobBuffers.push(blob); totalSize += 4 + blob.length; } - + // Phase 2: Single allocation and copy - const result = totalSize >= 1024 ? globalBufferPool.acquire(totalSize) : Buffer.allocUnsafe(totalSize); + const result = Buffer.allocUnsafe(totalSize); let offset = 0; - + for (const blob of blobBuffers) { result.writeInt32BE(blob.length, offset); offset += 4; blob.copy(result, offset); offset += blob.length; } - - return result.subarray(0, totalSize); + + return result; } /** * Serialize timestamps array to buffer (used by insertTablet) - * Optimized: Direct conversion to BigInt buffer + * Optimized: BigInt-free hi/lo writes (timestamps are safe integers) */ export function serializeTimestamps(timestamps: number[]): Buffer { - const size = timestamps.length * 8; - const buffer = size >= 1024 ? globalBufferPool.acquire(size) : Buffer.allocUnsafe(size); - + const buffer = Buffer.allocUnsafe(timestamps.length * 8); for (let i = 0; i < timestamps.length; i++) { const t = timestamps[i]; if (typeof t !== "number" || !Number.isFinite(t)) { throw new Error(`Invalid timestamp at index ${i}: ${t}`); } - buffer.writeBigInt64BE(BigInt(Math.floor(t)), i * 8); + const v = Math.floor(t); + buffer.writeInt32BE(Math.floor(v / 0x100000000), i * 8); + buffer.writeUInt32BE(v >>> 0, i * 8 + 4); } - - return buffer.subarray(0, size); + return buffer; } /** @@ -271,3 +283,248 @@ export function serializeColumnFast(values: any[], dataType: number): Buffer { throw new Error(`Unsupported data type: ${dataType}`); } } + +/** + * Serialize a whole tablet's values (all columns + null bitmaps) into ONE buffer. + * + * Wire format (identical to the legacy per-column path): + * [col0 data][col1 data]...[colN data] + * then per column: [hasNull flag byte][bitmap bytes, only when flag=1] + * Bitmap: ceil(rowCount/8) bytes, LSB-first, bit=1 means NULL. + * + * Optimizations over the legacy path: + * - No rows→columns transpose (reads values[row][col] directly) + * - No per-column intermediate buffers or boolean[] bitmaps + * - No trailing Buffer.concat (exact data size pre-computed; bitmap section + * allocated worst-case and trimmed with a zero-copy subarray) + * - Null bitmaps are packed inline during the value pass; the flag byte + * stays 0 and no bitmap bytes are emitted when a column has no nulls + * + * @param values Row-major tablet values: values[rowIndex][colIndex] + * @param dataTypes TSDataType code per column + * @param rowCount Number of rows + */ +export function serializeTabletValuesFast( + values: any[][], + dataTypes: number[], + rowCount: number, +): Buffer { + const numCols = dataTypes.length; + const bitmapBytes = Math.ceil(rowCount / 8); + + // ---- Pass 1: exact data-section size (fixed widths from schema; one + // byteLength scan for variable-width columns, memoizing repeated strings) ---- + let dataSize = 0; + for (let c = 0; c < numCols; c++) { + switch (dataTypes[c]) { + case 0: // BOOLEAN + dataSize += rowCount; + break; + case 1: // INT32 + case 3: // FLOAT + case 9: // DATE + dataSize += rowCount * 4; + break; + case 2: // INT64 + case 4: // DOUBLE + case 8: // TIMESTAMP + dataSize += rowCount * 8; + break; + case 5: // TEXT + case 11: { // STRING + let lastStr: string | null = null; + let lastLen = 0; + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + dataSize += 4; + continue; + } + const str = typeof v === "string" ? v : String(v); + if (str !== lastStr) { + lastLen = Buffer.byteLength(str, "utf8"); + lastStr = str; + } + dataSize += 4 + lastLen; + } + break; + } + case 10: { // BLOB + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + dataSize += 4 + (v === null || v === undefined + ? 0 + : Buffer.isBuffer(v) ? v.length : Buffer.from(v).length); + } + break; + } + default: + throw new Error(`Unsupported data type: ${dataTypes[c]}`); + } + } + + // Single allocation: data section + worst-case bitmap section + // (flag byte per column + bitmap per column); trimmed at the end. + const buffer = Buffer.allocUnsafe(dataSize + numCols * (1 + bitmapBytes)); + + // ---- Pass 2: write values and pack null bitmaps inline ---- + let off = 0; // data-section write offset + let bmOff = dataSize; // bitmap-section write offset (compact) + + for (let c = 0; c < numCols; c++) { + const flagPos = bmOff; + buffer[flagPos] = 0; // hasNull flag; flipped on first null + let hasNull = false; + + // Marks row r as NULL: lazily initializes this column's bitmap region. + const markNull = (r: number): void => { + if (!hasNull) { + hasNull = true; + buffer[flagPos] = 1; + buffer.fill(0, flagPos + 1, flagPos + 1 + bitmapBytes); + } + buffer[flagPos + 1 + (r >>> 3)] |= 1 << (r & 7); + }; + + switch (dataTypes[c]) { + case 0: // BOOLEAN + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer[off] = 0; + } else { + buffer[off] = v ? 1 : 0; + } + off += 1; + } + break; + case 1: // INT32 + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeInt32BE(0, off); + } else { + buffer.writeInt32BE(v, off); + } + off += 4; + } + break; + case 2: // INT64 + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeInt32BE(0, off); + buffer.writeUInt32BE(0, off + 4); + } else { + writeInt64BE(buffer, v, off); + } + off += 8; + } + break; + case 3: // FLOAT + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeFloatBE(0.0, off); + } else { + buffer.writeFloatBE(v, off); + } + off += 4; + } + break; + case 4: // DOUBLE + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeDoubleBE(0.0, off); + } else { + buffer.writeDoubleBE(v, off); + } + off += 8; + } + break; + case 5: // TEXT + case 11: { // STRING + // Single-entry encode cache: TAG columns repeat one string per tablet + let lastStr: string | null = null; + let lastBuf: Buffer | null = null; + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeInt32BE(0, off); + off += 4; + continue; + } + const str = typeof v === "string" ? v : String(v); + if (str !== lastStr || lastBuf === null) { + lastBuf = Buffer.from(str, "utf8"); + lastStr = str; + } + buffer.writeInt32BE(lastBuf.length, off); + off += 4; + lastBuf.copy(buffer, off); + off += lastBuf.length; + } + break; + } + case 8: // TIMESTAMP + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeInt32BE(0, off); + buffer.writeUInt32BE(0, off + 4); + } else { + writeInt64BE(buffer, v instanceof Date ? v.getTime() : v, off); + } + off += 8; + } + break; + case 9: // DATE + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeInt32BE(0, off); + } else { + const days = v instanceof Date + ? Math.floor(v.getTime() / (24 * 60 * 60 * 1000)) + : v; + buffer.writeInt32BE(days, off); + } + off += 4; + } + break; + case 10: { // BLOB + for (let r = 0; r < rowCount; r++) { + const v = values[r][c]; + if (v === null || v === undefined) { + markNull(r); + buffer.writeInt32BE(0, off); + off += 4; + continue; + } + const blob = Buffer.isBuffer(v) ? v : Buffer.from(v); + buffer.writeInt32BE(blob.length, off); + off += 4; + blob.copy(buffer, off); + off += blob.length; + } + break; + } + default: + throw new Error(`Unsupported data type: ${dataTypes[c]}`); + } + + bmOff = flagPos + 1 + (hasNull ? bitmapBytes : 0); + } + + // Zero-copy trim: unused worst-case bitmap space is dropped + return buffer.subarray(0, bmOff); +} diff --git a/tests/unit/FastSerializer.test.ts b/tests/unit/FastSerializer.test.ts index 4f2b7ba..3d9feba 100644 --- a/tests/unit/FastSerializer.test.ts +++ b/tests/unit/FastSerializer.test.ts @@ -29,6 +29,7 @@ import { serializeBlobColumn, serializeTimestamps, serializeColumnFast, + writeInt64BE, } from "../../src/utils/FastSerializer"; import { globalBufferPool } from "../../src/utils/BufferPool"; @@ -80,6 +81,110 @@ describe("FastSerializer", () => { }); }); + describe("BigInt-free INT64 hi/lo writes", () => { + const roundTrip = (v: number | bigint): bigint => { + const buf = Buffer.allocUnsafe(8); + writeInt64BE(buf, v, 0); + return buf.readBigInt64BE(0); + }; + + it("should write positive numbers correctly", () => { + expect(roundTrip(0)).toBe(BigInt(0)); + expect(roundTrip(1)).toBe(BigInt(1)); + expect(roundTrip(0xffffffff)).toBe(BigInt(0xffffffff)); // 32-bit boundary + expect(roundTrip(0x100000000)).toBe(BigInt("4294967296")); + expect(roundTrip(Number.MAX_SAFE_INTEGER)).toBe(BigInt(Number.MAX_SAFE_INTEGER)); + }); + + it("should write negative numbers correctly (two's complement)", () => { + expect(roundTrip(-1)).toBe(BigInt(-1)); + expect(roundTrip(-2000)).toBe(BigInt(-2000)); + expect(roundTrip(-0x100000000)).toBe(BigInt("-4294967296")); + expect(roundTrip(-0x100000001)).toBe(BigInt("-4294967297")); + expect(roundTrip(Number.MIN_SAFE_INTEGER)).toBe(BigInt(Number.MIN_SAFE_INTEGER)); + }); + + it("should handle BigInt inputs", () => { + expect(roundTrip(BigInt(1000))).toBe(BigInt(1000)); + expect(roundTrip(BigInt(-2000))).toBe(BigInt(-2000)); + expect(roundTrip(BigInt("-9223372036854775808"))).toBe(BigInt("-9223372036854775808")); // i64 min + expect(roundTrip(BigInt("9223372036854775807"))).toBe(BigInt("9223372036854775807")); // i64 max + }); + + it("should match legacy BigInt output for INT64 columns with negatives", () => { + const values = [ + Number.MIN_SAFE_INTEGER, + -1, + -0x100000000, + 0, + 0xffffffff, + Number.MAX_SAFE_INTEGER, + BigInt(-42), + ]; + const buffer = serializeInt64Column(values); + const legacy = Buffer.alloc(values.length * 8); + values.forEach((v, i) => legacy.writeBigInt64BE(BigInt(v), i * 8)); + expect(buffer.equals(legacy)).toBe(true); + }); + + it("should serialize negative timestamps (pre-1970) correctly", () => { + const values = [-1, -86400000, 0, 1700000000000]; + const buffer = serializeTimestampColumn(values); + values.forEach((v, i) => { + expect(buffer.readBigInt64BE(i * 8)).toBe(BigInt(v)); + }); + // serializeTimestamps (time column) too + const tsBuffer = serializeTimestamps(values as number[]); + values.forEach((v, i) => { + expect(tsBuffer.readBigInt64BE(i * 8)).toBe(BigInt(v)); + }); + }); + }); + + describe("String encode cache", () => { + it("should reuse encoding for repeated strings (TAG-like column)", () => { + const values = Array.from({ length: 100 }, () => "device_001"); + const buffer = serializeTextColumn(values); + let offset = 0; + for (let i = 0; i < 100; i++) { + const len = buffer.readInt32BE(offset); + offset += 4; + expect(buffer.toString("utf8", offset, offset + len)).toBe("device_001"); + offset += len; + } + expect(offset).toBe(buffer.length); + }); + + it("should not wrongly reuse cache for different strings", () => { + const values = ["aaa", "aaa", "bbb", "aaa", "cc", "你好", "你好", "cc"]; + const buffer = serializeTextColumn(values); + let offset = 0; + for (const v of values) { + const expected = Buffer.from(v, "utf8"); + const len = buffer.readInt32BE(offset); + offset += 4; + expect(len).toBe(expected.length); + expect(buffer.toString("utf8", offset, offset + len)).toBe(v); + offset += len; + } + expect(offset).toBe(buffer.length); + }); + + it("should handle nulls interleaved with cached strings", () => { + const values = ["x", null, "x", undefined, "y"]; + const buffer = serializeTextColumn(values); + let offset = 0; + const expected = ["x", "", "x", "", "y"]; + for (const v of expected) { + const len = buffer.readInt32BE(offset); + offset += 4; + expect(buffer.toString("utf8", offset, offset + len)).toBe(v); + offset += len; + } + expect(offset).toBe(buffer.length); + }); + }); + describe("FLOAT Serialization", () => { it("should serialize FLOAT values correctly", () => { const values = [1.5, -2.5, 0.0, null, undefined]; @@ -260,20 +365,20 @@ describe("FastSerializer", () => { }); describe("Buffer Pool Integration", () => { - it("should track buffer pool statistics", () => { + it("should no longer use the deprecated buffer pool", () => { const statsInitial = globalBufferPool.getStats(); - - // Allocate buffers larger than 1KB to trigger pooling + + // Large buffers used to trigger pooling; serializers now allocate directly const largeValues = Array.from({ length: 300 }, (_, i) => i); // 300 * 4 = 1200 bytes - serializeInt32Column(largeValues); // This should use the pool - serializeDoubleColumn(largeValues); // 300 * 8 = 2400 bytes, uses pool - + serializeInt32Column(largeValues); + serializeDoubleColumn(largeValues); // 300 * 8 = 2400 bytes + const statsFinal = globalBufferPool.getStats(); - - // Should have some allocations (or hits if pool was already populated) - const totalActivity = statsFinal.allocations + statsFinal.hits; - const initialActivity = statsInitial.allocations + statsInitial.hits; - expect(totalActivity).toBeGreaterThan(initialActivity); + + // The pool is deprecated: no acquire() calls should be made + expect(statsFinal.allocations + statsFinal.hits).toBe( + statsInitial.allocations + statsInitial.hits, + ); }); }); }); diff --git a/tests/unit/TabletSerialization.test.ts b/tests/unit/TabletSerialization.test.ts index fb86706..38ce0ba 100644 --- a/tests/unit/TabletSerialization.test.ts +++ b/tests/unit/TabletSerialization.test.ts @@ -288,6 +288,91 @@ describe('Tablet Serialization', () => { }); }); + describe('Fast vs Legacy Tablet Serialization (golden wire-format test)', () => { + let legacySession: Session; + + beforeEach(() => { + legacySession = new Session({ + host: 'localhost', + port: 6667, + username: 'root', + password: 'root', + enableFastSerialization: false, + }); + }); + + const compare = (values: any[][], dataTypes: number[], rowCount: number) => { + const fast: Buffer = (session as any).serializeTabletValues(values, dataTypes, rowCount); + const legacy: Buffer = (legacySession as any).serializeTabletValues(values, dataTypes, rowCount); + expect(fast.length).toBe(legacy.length); + expect(fast.equals(legacy)).toBe(true); + }; + + test('mixed-type tablet with nulls matches legacy byte-for-byte', () => { + // 10 rows × 8 columns covering every type, nulls scattered + const dataTypes = [ + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.TEXT, + TSDataType.TIMESTAMP, + TSDataType.STRING, + ]; + const values: any[][] = []; + for (let r = 0; r < 10; r++) { + values.push([ + r % 3 === 0 ? null : r % 2 === 0, + r === 1 ? null : r * 100 - 500, + r === 4 ? null : BigInt(-1000000000000) + BigInt(r), + r === 7 ? undefined : r * 1.5 - 3, + r === 2 ? null : r * 2.5e10, + r === 5 ? null : `tag_${r % 2}`, // repeated strings exercise the cache + r === 8 ? null : 1700000000000 + r, + r === 0 ? null : '设备_' + (r % 3), + ]); + } + compare(values, dataTypes, 10); + }); + + test('tablet with no nulls matches legacy byte-for-byte', () => { + const dataTypes = [TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT]; + const values: any[][] = []; + for (let r = 0; r < 17; r++) { + // 17 rows: bitmap byte-count edge (ceil(17/8)=3) + values.push([r * 1.1, -r * 12345678901, 'constant_tag']); + } + compare(values, dataTypes, 17); + }); + + test('tablet with all-null column matches legacy byte-for-byte', () => { + const dataTypes = [TSDataType.INT32, TSDataType.TEXT, TSDataType.BOOLEAN]; + const values: any[][] = []; + for (let r = 0; r < 9; r++) { + values.push([null, r % 2 === 0 ? 's' : null, true]); + } + compare(values, dataTypes, 9); + }); + + test('BLOB and DATE tablet matches legacy byte-for-byte', () => { + const dataTypes = [TSDataType.BLOB, TSDataType.DATE]; + const values: any[][] = [ + [Buffer.from([1, 2, 3]), new Date('2024-01-01')], + [null, 19722], + [Buffer.alloc(0), null], + [Buffer.from([0xff]), new Date('2025-06-15')], + ]; + compare(values, dataTypes, 4); + }); + + test('single-row and empty-ish edge cases match legacy', () => { + compare([[42]], [TSDataType.INT32], 1); + compare([[null]], [TSDataType.DOUBLE], 1); + compare([[9007199254740991, -9007199254740991]].map((r) => r), [TSDataType.INT64, TSDataType.INT64], 1); + }); + }); + describe('Column Serialization - Other Data Types', () => { test('should serialize BOOLEAN column', () => { const values = [true, false, true, false];
