This is an automated email from the ASF dual-hosted git repository. CritasWang pushed a commit to branch feat/tablet-object-write in repository https://gitbox.apache.org/repos/asf/iotdb-client-nodejs.git
commit 9b4d18e380d8fd04e5b570413e41b4b90b233dc2 Author: CritasWang <[email protected]> AuthorDate: Wed Aug 26 09:56:01 2026 +0800 feat: support OBJECT data type in table-model tablet writes Add TSDataType.OBJECT (12) and table-model OBJECT write support: - TableTablet.buildObjectValue / setObjectValueAt with the Java-compatible 9-byte segment header (isEOF + big-endian offset + content) - Serialize OBJECT columns as BLOB-like length-prefixed binary in both the legacy and fast tablet serialization paths - Decode OBJECT columns from TsBlock results as "(Object) ..." size summaries - Add mock/unit tests for segment framing, RPC request bytes, null bitmap, serialization parity, and TsBlock decoding Refs apache/iotdb-client-go#175 --- src/client/ColumnDecoder.ts | 12 ++- src/client/Session.ts | 96 +++++++++++++++--- src/utils/DataTypes.ts | 47 ++++++++- src/utils/FastSerializer.ts | 7 +- tests/unit/DataTypes.test.ts | 48 +++++++++ tests/unit/TableTabletObject.test.ts | 171 +++++++++++++++++++++++++++++++++ tests/unit/TabletSerialization.test.ts | 30 ++++++ 7 files changed, 392 insertions(+), 19 deletions(-) diff --git a/src/client/ColumnDecoder.ts b/src/client/ColumnDecoder.ts index 25d5c13..f92ff3d 100644 --- a/src/client/ColumnDecoder.ts +++ b/src/client/ColumnDecoder.ts @@ -17,8 +17,7 @@ * under the License. */ -import { logger } from "../utils/Logger"; -import { parseIntToDate } from "../utils/DataTypes"; +import { objectBytesToString, parseIntToDate } from "../utils/DataTypes"; /** * Column encoding types matching Apache IoTDB ColumnEncoding enum @@ -323,7 +322,7 @@ class ByteArrayColumnDecoder implements ColumnDecoder { /** * Decoder for BINARY array encoding (encoding=3) - * Handles TEXT, STRING, and BLOB data types with variable-length values + * Handles TEXT, STRING, BLOB, and OBJECT data types with variable-length values */ class BinaryArrayColumnDecoder implements ColumnDecoder { readColumn( @@ -332,8 +331,8 @@ class BinaryArrayColumnDecoder implements ColumnDecoder { dataType: number, positionCount: number, ): { column: Column; bytesRead: number } { - // Supports TEXT(5), BLOB(10), STRING(11) - if (dataType !== 5 && dataType !== 10 && dataType !== 11) { + // Supports TEXT(5), BLOB(10), STRING(11), OBJECT(12) + if (dataType !== 5 && dataType !== 10 && dataType !== 11 && dataType !== 12) { throw new Error( `Invalid data type ${dataType} for BinaryArrayColumnDecoder`, ); @@ -370,6 +369,9 @@ class BinaryArrayColumnDecoder implements ColumnDecoder { if (dataType === 10) { // BLOB - keep as Buffer values[i] = data; + } else if (dataType === 12) { + // OBJECT - render as "(Object) 1.00 KB" like the Go/C# clients + values[i] = objectBytesToString(data); } else { // TEXT/STRING - convert to UTF-8 string values[i] = data.toString("utf8"); diff --git a/src/client/Session.ts b/src/client/Session.ts index ef6424a..e65d627 100644 --- a/src/client/Session.ts +++ b/src/client/Session.ts @@ -29,12 +29,12 @@ import { registerClosable, unregisterClosable } from "../utils/ProcessCleanup"; import { SessionDataSet } from "./SessionDataSet"; import { RowRecord } from "./RowRecord"; import { BaseColumnDecoder, ColumnEncoding, Column } from "./ColumnDecoder"; -import { RedirectException, isWildcardAddress } from "../utils/Errors"; +import { isWildcardAddress } from "../utils/Errors"; import { serializeTabletValuesFast, serializeTimestamps } from "../utils/FastSerializer"; -import { parseDateToInt, parseIntToDate } from "../utils/DataTypes"; +import { parseDateToInt, parseIntToDate, TSDataType } from "../utils/DataTypes"; const ttypes = require("../thrift/generated/client_types"); @@ -172,6 +172,75 @@ export class TableTablet implements ITableTablet { this.timestamps.push(timestamp); this.values.push(values); } + + /** + * Build the wire representation of one OBJECT segment: a 1-byte isEOF + * flag, an 8-byte big-endian offset, then the raw segment content. + * This matches Java's + * Tablet.addValue(rowIndex, columnIndex, isEOF, offset, content). + * + * @param isEOF - Whether this segment is the last one of the object + * @param offset - Byte offset of this segment within the whole object + * @param content - Raw bytes of this segment + */ + static buildObjectValue( + isEOF: boolean, + offset: number | bigint, + content: Buffer | Uint8Array, + ): Buffer { + let bigintOffset = typeof offset === "bigint" ? offset : BigInt(offset); + if ( + typeof offset === "number" && + (!Number.isSafeInteger(offset) || offset < 0) + ) { + throw new Error(`Invalid OBJECT segment offset: ${offset}`); + } + if (bigintOffset < 0n) { + throw new Error(`Invalid OBJECT segment offset: ${offset}`); + } + const raw = Buffer.isBuffer(content) ? content : Buffer.from(content); + const value = Buffer.allocUnsafe(9 + raw.length); + value[0] = isEOF ? 1 : 0; + value.writeBigUInt64BE(bigintOffset, 1); + raw.copy(value, 9); + return value; + } + + /** + * Write one segment of an OBJECT column value at an existing row. + * + * An OBJECT value can be written in multiple segments so a large object + * does not need to be fully loaded into memory. Segments must be written + * with ascending offsets and the last segment must set isEOF to true. + * + * @param isEOF - Whether this segment is the last one of the object + * @param offset - Byte offset of this segment within the whole object + * @param content - Raw bytes of this segment + * @param columnIndex - Index of the OBJECT column + * @param rowIndex - Index of the row to write into + */ + setObjectValueAt( + isEOF: boolean, + offset: number | bigint, + content: Buffer | Uint8Array, + columnIndex: number, + rowIndex: number, + ): void { + if (columnIndex < 0 || columnIndex >= this.columnTypes.length) { + throw new Error(`Illegal columnIndex: ${columnIndex}`); + } + if (this.columnTypes[columnIndex] !== TSDataType.OBJECT) { + throw new Error(`Column ${columnIndex} must be of type OBJECT`); + } + if (rowIndex < 0 || rowIndex >= this.values.length) { + throw new Error(`Illegal rowIndex: ${rowIndex}`); + } + this.values[rowIndex][columnIndex] = TableTablet.buildObjectValue( + isEOF, + offset, + content, + ); + } } /** @@ -760,8 +829,9 @@ export class Session { }); return buffer; } - case 10: { - // BLOB + case 10: // BLOB + case 12: { + // OBJECT (table model) uses the same binary length-prefix encoding as BLOB // Optimized: Pre-calculate total size to avoid multiple Buffer.concat calls // Phase 1: Convert all values to buffers and calculate total size @@ -1152,6 +1222,7 @@ export class Session { else if (type.includes("DATE")) return 9; else if (type.includes("BLOB")) return 10; else if (type.includes("STRING")) return 11; + else if (type.includes("OBJECT")) return 12; return 5; // Default to TEXT } @@ -1178,15 +1249,18 @@ export class Session { case 5: // TEXT - variable length, need to parse case 10: // BLOB - variable length case 11: // STRING - variable length + case 12: // OBJECT - variable length (binary, length-prefixed) // For variable-length types, count entries by parsing length prefixes - let count = 0; - let offset = 0; - while (offset + 4 <= length) { - const strLength = buffer.readInt32BE(offset); - offset += 4 + strLength; - count++; + { + let count = 0; + let offset = 0; + while (offset + 4 <= length) { + const strLength = buffer.readInt32BE(offset); + offset += 4 + strLength; + count++; + } + return count; } - return count; default: logger.warn( `Unknown data type ${dataType}, cannot determine row count`, diff --git a/src/utils/DataTypes.ts b/src/utils/DataTypes.ts index 6fd2de3..fde8785 100644 --- a/src/utils/DataTypes.ts +++ b/src/utils/DataTypes.ts @@ -96,7 +96,13 @@ export enum TSDataType { */ STRING = 11, - // OBJECT = 12, // Reserved - not yet implemented + /** + * OBJECT (table model only) + * JavaScript type: Buffer (built with the segment framing: 1-byte isEOF + + * 8-byte big-endian offset + raw content) + * Storage size: variable (4-byte length prefix + binary content) + */ + OBJECT = 12, } /** @@ -241,7 +247,46 @@ export function getDataTypeName(typeCode: number): string { return "BLOB"; case TSDataType.STRING: return "STRING"; + case TSDataType.OBJECT: + return "OBJECT"; default: return "UNKNOWN"; } } + +/** + * Format the wire representation of a stored OBJECT value for display. + * + * The server stores OBJECT cells as an 8-byte big-endian file size followed + * by the internal object path. Mirroring the Go client's objectBytesToString, + * this helper renders the size in human-readable units: + * "(Object) 1023 B", "(Object) 1.00 KB", "(Object) 1.00 MB", + * "(Object) 1.00 GB". + * + * @param input - Raw OBJECT cell bytes (at least the 8-byte size prefix) + * @returns Human-readable size string + * @throws Error if the input is shorter than 8 bytes + */ +export function objectBytesToString(input: Buffer | Uint8Array): string { + if (input.length < 8) { + throw new Error( + "Invalid OBJECT value: expected at least 8 bytes, got " + input.length, + ); + } + const buffer = Buffer.isBuffer(input) ? input : Buffer.from(input); + const size = buffer.readUInt32BE(0) * 0x100000000 + buffer.readUInt32BE(4); + const KILOBYTE = 1024; + const MEGABYTE = KILOBYTE * 1024; + const GIGABYTE = MEGABYTE * 1024; + + if (size < KILOBYTE) { + return "(Object) " + size + " B"; + } + if (size < MEGABYTE) { + return "(Object) " + (size / KILOBYTE).toFixed(2) + " KB"; + } + if (size < GIGABYTE) { + return "(Object) " + (size / MEGABYTE).toFixed(2) + " MB"; + } + return "(Object) " + (size / GIGABYTE).toFixed(2) + " GB"; +} diff --git a/src/utils/FastSerializer.ts b/src/utils/FastSerializer.ts index f0d45f5..bbff140 100644 --- a/src/utils/FastSerializer.ts +++ b/src/utils/FastSerializer.ts @@ -299,6 +299,7 @@ export function serializeColumnFast(values: any[], dataType: number): Buffer { case 9: // DATE return serializeDateColumn(values); case 10: // BLOB + case 12: // OBJECT (same length-prefixed binary encoding) return serializeBlobColumn(values); default: throw new Error(`Unsupported data type: ${dataType}`); @@ -370,7 +371,8 @@ export function serializeTabletValuesFast( } break; } - case 10: { // BLOB + case 10: // BLOB + case 12: { // OBJECT (same length-prefixed binary encoding) for (let r = 0; r < rowCount; r++) { const v = values[r][c]; dataSize += 4 + (v === null || v === undefined ? 0 : blobByteLength(v)); @@ -517,7 +519,8 @@ export function serializeTabletValuesFast( off += 4; } break; - case 10: { // BLOB + case 10: // BLOB + case 12: { // OBJECT (same length-prefixed binary encoding) for (let r = 0; r < rowCount; r++) { const v = values[r][c]; if (v === null || v === undefined) { diff --git a/tests/unit/DataTypes.test.ts b/tests/unit/DataTypes.test.ts index fd300a2..359a4b7 100644 --- a/tests/unit/DataTypes.test.ts +++ b/tests/unit/DataTypes.test.ts @@ -18,8 +18,11 @@ */ import { + getDataTypeName, + objectBytesToString, parseDateToInt, parseIntToDate, + TSDataType, } from "../../src/utils/DataTypes"; import { BaseColumnDecoder, @@ -132,6 +135,51 @@ describe("DATE yyyyMMdd conversion", () => { }); }); + describe("OBJECT type codes and display formatting", () => { + it("exposes OBJECT as TSDataType 12 and resolves its name", () => { + expect(TSDataType.OBJECT).toBe(12); + expect(getDataTypeName(12)).toBe("OBJECT"); + }); + + it.each([ + [1023, "(Object) 1023 B"], + [1024, "(Object) 1.00 KB"], + [1024 * 1024, "(Object) 1.00 MB"], + [1024 * 1024 * 1024, "(Object) 1.00 GB"], + ] as const)("formats %d bytes as %s", (size, expected) => { + const buffer = Buffer.alloc(8 + 4); + buffer.writeUInt32BE(Math.floor(size / 0x100000000), 0); + buffer.writeUInt32BE(size >>> 0, 4); + expect(objectBytesToString(buffer)).toBe(expected); + }); + + it("rejects OBJECT values shorter than the 8-byte size prefix", () => { + expect(() => objectBytesToString(Buffer.alloc(7))).toThrow( + /expected at least 8 bytes/, + ); + }); + }); + + describe("TsBlock column decode (BinaryArrayColumnDecoder for OBJECT)", () => { + it("decodes an OBJECT cell as a human-readable size string", () => { + // BinaryArray column: mayHaveNull=0, length i32 BE, 8-byte size + path. + const payload = Buffer.alloc(8 + "internal/path/1.bin".length); + payload.writeUInt32BE(0, 0); + payload.writeUInt32BE(1024, 4); + payload.write("internal/path/1.bin", 8, "utf8"); + + const buffer = Buffer.concat([ + Buffer.from([0x00]), // mayHaveNull + Buffer.from([0, 0, 0, payload.length]), + payload, + ]); + + const decoder = BaseColumnDecoder.getDecoder(ColumnEncoding.BinaryArray); + const { column } = decoder.readColumn(buffer, 0, TSDataType.OBJECT, 1); + expect(column.values[0]).toBe("(Object) 1.00 KB"); + }); + }); + describe("TsBlock column decode (Int32ArrayColumnDecoder)", () => { it("should decode a DATE column value as a Date from yyyyMMdd wire bytes", () => { // Column layout: 1 byte null flag (0 = no nulls) + INT32 BE values diff --git a/tests/unit/TableTabletObject.test.ts b/tests/unit/TableTabletObject.test.ts new file mode 100644 index 0000000..1ca08e1 --- /dev/null +++ b/tests/unit/TableTabletObject.test.ts @@ -0,0 +1,171 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { ColumnCategory, Session, TableTablet } from "../../src/client/Session"; +import { TSDataType } from "../../src/utils/DataTypes"; + +describe("TableTablet OBJECT write support", () => { + const newObjectTablet = () => + new TableTablet( + "object_table", + ["region_id", "file"], + [TSDataType.STRING, TSDataType.OBJECT], + [ColumnCategory.TAG, ColumnCategory.FIELD], + ); + + describe("TableTablet.buildObjectValue", () => { + it("builds the Java-compatible segment frame (isEOF + offset + content)", () => { + expect(TableTablet.buildObjectValue(true, 0, Buffer.from([0x11, 0x22]))).toEqual( + Buffer.from([1, 0, 0, 0, 0, 0, 0, 0, 0, 0x11, 0x22]), + ); + expect( + TableTablet.buildObjectValue(false, 512, Buffer.from([0x33])), + ).toEqual( + Buffer.from([0, 0, 0, 0, 0, 0, 0, 2, 0, 0x33]), + ); + }); + + it("accepts Uint8Array content and bigint offsets", () => { + const value = TableTablet.buildObjectValue( + true, + BigInt(512), + new Uint8Array([0x33]), + ); + expect(value).toEqual(Buffer.from([1, 0, 0, 0, 0, 0, 0, 2, 0, 0x33])); + }); + + it("rejects negative and unsafe offsets", () => { + expect(() => TableTablet.buildObjectValue(true, -1, Buffer.alloc(0))).toThrow( + /offset/, + ); + expect(() => + TableTablet.buildObjectValue(true, 2 ** 53, Buffer.alloc(0)), + ).toThrow(/offset/); + }); + }); + + describe("TableTablet.setObjectValueAt", () => { + it("writes whole-object and segmented values at the requested row/column", () => { + const tablet = newObjectTablet(); + tablet.addRow(1, ["r1", null]); + tablet.addRow(2, ["r2", null]); + + tablet.setObjectValueAt(true, 0, Buffer.from([0x11, 0x22]), 1, 0); + tablet.setObjectValueAt(true, 512, Buffer.from([0x33]), 1, 1); + + expect(tablet.values[0][1]).toEqual( + Buffer.from([1, 0, 0, 0, 0, 0, 0, 0, 0, 0x11, 0x22]), + ); + expect(tablet.values[1][1]).toEqual( + Buffer.from([1, 0, 0, 0, 0, 0, 0, 2, 0, 0x33]), + ); + }); + + it("rejects non-OBJECT columns and out-of-range indexes", () => { + const tablet = newObjectTablet(); + tablet.addRow(1, ["r1", null]); + + expect(() => + tablet.setObjectValueAt(true, 0, Buffer.from([0x01]), 0, 0), + ).toThrow(/must be of type OBJECT/); + expect(() => + tablet.setObjectValueAt(true, 0, Buffer.from([0x01]), 1, -1), + ).toThrow(/rowIndex/); + expect(() => + tablet.setObjectValueAt(true, 0, Buffer.from([0x01]), -1, 0), + ).toThrow(/columnIndex/); + expect(() => + tablet.setObjectValueAt(true, 0, Buffer.from([0x01]), 1, 1), + ).toThrow(/rowIndex/); + }); + }); + + describe("Session.insertTablet with a mocked Thrift client", () => { + let session: Session; + let capturedReq: any; + let insertTablet: jest.Mock; + + beforeEach(() => { + session = new Session({ + host: "localhost", + port: 6667, + username: "root", + password: "root", + }); + insertTablet = jest.fn((_req: unknown, callback: (err: null, response: any) => void) => { + capturedReq = _req; + callback(null, { code: 200 }); + }); + (session as any).connection = { + getClient: () => ({ insertTablet }), + getSessionId: () => 1, + }; + }); + + it("sends TSDataType 12 and the binary segment payload to the server", async () => { + const tablet = newObjectTablet(); + tablet.addRow(1608268702780, ["r1", null]); + tablet.setObjectValueAt( + true, + 0, + Buffer.from([0x01, 0x02, 0x03]), + 1, + 0, + ); + + await session.insertTablet(tablet); + + expect(insertTablet).toHaveBeenCalledTimes(1); + expect(capturedReq.writeToTable).toBe(true); + expect(capturedReq.types).toEqual([TSDataType.STRING, TSDataType.OBJECT]); + expect(capturedReq.size).toBe(1); + const expectedTimestamps = Buffer.alloc(8); + expectedTimestamps.writeBigInt64BE(BigInt(1608268702780), 0); + expect(capturedReq.timestamps.equals(expectedTimestamps)).toBe(true); + + // STRING 'r1': i32 len 2 + 'r1'; OBJECT segment: i32 len 12 + payload; + // two no-null bitmap flags. + const expectedValues = Buffer.concat([ + Buffer.from([0, 0, 0, 2, 0x72, 0x31]), + Buffer.from([0, 0, 0, 12]), + Buffer.from([1, 0, 0, 0, 0, 0, 0, 0, 0, 0x01, 0x02, 0x03]), + Buffer.from([0, 0]), + ]); + expect(capturedReq.values.equals(expectedValues)).toBe(true); + }); + + it("marks a null OBJECT cell in the bitmap while keeping the payload empty", async () => { + const tablet = newObjectTablet(); + tablet.addRow(1, ["r1", null]); + + await session.insertTablet(tablet); + + const values: Buffer = capturedReq.values; + // STRING: 6 bytes ('r1') + OBJECT empty: 4 bytes + bitmap section: + // col0 flag 0, col1 flag 1 + 1 bitmap byte (bit 0 set for row 0). + expect(values.length).toBe(6 + 4 + 1 + 1 + 1); + const expected = Buffer.concat([ + Buffer.from([0, 0, 0, 2, 0x72, 0x31]), + Buffer.from([0, 0, 0, 0]), + Buffer.from([0x00, 0x01, 0x01]), + ]); + expect(values.equals(expected)).toBe(true); + }); + }); +}); diff --git a/tests/unit/TabletSerialization.test.ts b/tests/unit/TabletSerialization.test.ts index a3ddfa8..832e791 100644 --- a/tests/unit/TabletSerialization.test.ts +++ b/tests/unit/TabletSerialization.test.ts @@ -286,6 +286,26 @@ describe('Tablet Serialization', () => { expect(buffer.readInt32BE(offset)).toBe(1); }); + + test('should serialize OBJECT column with the same length-prefix encoding as BLOB', () => { + const segment1 = Buffer.from([1, 0, 0, 0, 0, 0, 0, 0, 0, 0x11, 0x22]); + const segment2 = Buffer.from([1, 0, 0, 0, 0, 0, 0, 2, 0, 0x33]); + const values = [segment1, null, segment2]; + const buffer = (session as any).serializeColumn(values, TSDataType.OBJECT); + + let offset = 0; + expect(buffer.readInt32BE(offset)).toBe(11); + offset += 4; + expect(buffer.subarray(offset, offset + 11).equals(segment1)).toBe(true); + offset += 11; + + expect(buffer.readInt32BE(offset)).toBe(0); // null → empty + offset += 4; + + expect(buffer.readInt32BE(offset)).toBe(10); + offset += 4; + expect(buffer.subarray(offset, offset + 10).equals(segment2)).toBe(true); + }); }); describe('Fast vs Legacy Tablet Serialization (golden wire-format test)', () => { @@ -366,6 +386,16 @@ describe('Tablet Serialization', () => { compare(values, dataTypes, 4); }); + test('OBJECT tablet matches legacy byte-for-byte', () => { + const dataTypes = [TSDataType.STRING, TSDataType.OBJECT]; + const values: any[][] = [ + ['tag-1', Buffer.from([1, 0, 0, 0, 0, 0, 0, 0, 0, 0x11, 0x22])], + ['tag-2', null], + ['tag-3', Buffer.from([1, 0, 0, 0, 0, 0, 0, 2, 0, 0x33])], + ]; + compare(values, dataTypes, 3); + }); + test('non-Buffer BLOB inputs (Uint8Array, byte array, string) match legacy', () => { const dataTypes = [TSDataType.BLOB]; const values: any[][] = [
