Feat: supports decimal type

nodejs/src/client/wsClient.ts

@@ -61,7 +61,6 @@ export class WsClient {
             throw(new TDWebSocketClientError(ErrorCode.ERR_WEBSOCKET_CONNECTION_FAIL, `connection creation failed, url: ${this._url}, code:${e.code}, msg:${e.message}`));
         }
 
-
     }
 
     async setOptionConnection(option: TSDB_OPTION_CONNECTION, value: string | null): Promise<void> {

nodejs/src/client/wsResponse.ts

@@ -33,6 +33,8 @@ export class WSQueryResponse {
     fields_names?: Array<string> | null;
     fields_types?: Array<number> | null;
     fields_lengths?: Array<number> | null;
+    fields_precisions?: Array<bigint> | null;
+    fields_scales?: Array<bigint> | null;
     precision?: number;
 
     constructor(resp:MessageResp) {
@@ -58,6 +60,8 @@ export class WSQueryResponse {
         this.fields_types = msg.fields_types;
         this.fields_lengths = msg.fields_lengths;
         this.precision = msg.precision;
+        this.fields_precisions = msg.fields_precisions ? msg.fields_precisions.map((p: number) => BigInt(p)) : [];
+        this.fields_scales = msg.fields_scales ? msg.fields_scales.map((s: number) => BigInt(s)) : [];
     }
 }
 

nodejs/src/common/constant.ts

@@ -58,7 +58,9 @@ export enum TDengineTypeCode {
     BIGINT_UNSIGNED = 14,
     JSON = 15,
     VARBINARY = 16,
+    DECIMAL = 17,
     GEOMETRY = 20,
+    DECIMAL64 = 21,
 }
 
 export enum TSDB_OPTION_CONNECTION {

nodejs/src/common/taosResult.ts

@@ -3,6 +3,8 @@ import { ColumnsBlockType, TDengineTypeCode, TDengineTypeName } from './constant
 import { ErrorCode, TaosResultError, WebSocketQueryInterFaceError } from "./wsError";
 import { appendRune } from "./ut8Helper"
 import logger from "./log";
+import { decimalToString } from "./utils";
+import { TMQRawDataSchema } from "../tmq/constant";
 
 export interface TDengineMeta {
     name: string,
@@ -29,7 +31,9 @@ export class TaosResult {
     private _precision: number | null | undefined;
     protected _affectRows: number | null | undefined;
     private _totalTime = 0;
-
+    private fields_precisions?: Array<bigint> | null;
+    private fields_scales?: Array<bigint> | null;
+
     /** unit nano seconds */
     private _timing: bigint | null | undefined;
     constructor(queryResponse?: WSQueryResponse) {
@@ -65,6 +69,8 @@ export class TaosResult {
         this._timing = queryResponse.timing
         this._precision = queryResponse.precision
         this._totalTime = queryResponse.totalTime
+        this.fields_precisions = queryResponse.fields_precisions
+        this.fields_scales = queryResponse.fields_scales
     }
 
     public setPrecision(precision: number) {
@@ -134,6 +140,14 @@ export class TaosResult {
             this._timing = this._timing + timing
         }    
     }
+
+    public getFieldsScales(index: number): bigint | null {
+        if (this.fields_scales) {
+            return this.fields_scales[index];
+        }
+        return null;
+    }
+
     /**
      * Mapping the WebSocket response type code to TDengine's type name. 
      */
@@ -201,7 +215,7 @@ export function parseBlock(blocks: WSFetchBlockResponse, taosResult: TaosResult)
                     if (bitFlag == 1) {
                         row.push("NULL")
                     } else {
-                        row.push(readSolidData(dataView, colDataHead, metaList[j]))
+                        row.push(readSolidData(dataView, colDataHead, metaList[j], taosResult.getFieldsScales(j)));
                     }
 
                     colBlockHead = colBlockHead + bitMapSize + dataView.getInt32(INT_32_SIZE * j, true)
@@ -261,7 +275,7 @@ export function _isVarType(metaType: number): Number {
     }
 }
 export function readSolidDataToArray(dataBuffer: DataView, colBlockHead:number, 
-    rows:number, metaType: number, bitMapArr: Uint8Array): any[] {
+    rows:number, metaType: number, bitMapArr: Uint8Array, startOffset: number, colIndex: number): any[] {
 
     let result:any[] = []
     switch (metaType) {
@@ -369,14 +383,40 @@ export function readSolidDataToArray(dataBuffer: DataView, colBlockHead:number,
             }
             break;
         }
+        case TDengineTypeCode.DECIMAL64: {
+            let scale = getScaleFromRowBlock(dataBuffer, colIndex, startOffset);
+            for (let i = 0; i < rows; i++, colBlockHead += 8) {
+                if (isNull(bitMapArr, i)) {
+                    result.push(null);
+                }else{
+                    let decimalVal = dataBuffer.getBigInt64(colBlockHead, true)
+                    result.push(decimalToString(decimalVal.toString(), BigInt(scale)));
+                }
+            }
+            break;
+        }
+        case TDengineTypeCode.DECIMAL: {
+            let scale = getScaleFromRowBlock(dataBuffer, colIndex, startOffset);
+            for (let i = 0; i < rows; i++, colBlockHead += 16) {
+                if (isNull(bitMapArr, i)) {
+                    result.push(null);
+                }else{
+                    let decimalHighPart = dataBuffer.getBigInt64(colBlockHead + 8, true);
+                    const  decimalLowPart = dataBuffer.getBigUint64(colBlockHead, true);
+                    const decimalCombined = (decimalHighPart << 64n) | decimalLowPart;
+                    result.push(decimalToString(decimalCombined.toString(), BigInt(scale)));
+                }
+            }
+            break;
+        }
         default: {
-            throw new WebSocketQueryInterFaceError(ErrorCode.ERR_UNSUPPORTED_TDENGINE_TYPE, `unspported type ${metaType}`)
+            throw new WebSocketQueryInterFaceError(ErrorCode.ERR_UNSUPPORTED_TDENGINE_TYPE, `unsupported type ${metaType}`)
         }
     }
     return result;
 
 }
-export function readSolidData(dataBuffer: DataView, colDataHead: number, meta: ResponseMeta): Number | Boolean | BigInt {
+export function readSolidData(dataBuffer: DataView, colDataHead: number, meta: ResponseMeta, fields_scale: bigint | null): Number | Boolean | BigInt | string {
 
     switch (meta.type) {
         case TDengineTypeCode.BOOL: {
@@ -416,8 +456,18 @@ export function readSolidData(dataBuffer: DataView, colDataHead: number, meta: R
             return dataBuffer.getBigInt64(colDataHead, true);
             // could change 
         }
+        case TDengineTypeCode.DECIMAL: {
+            let decimalHighPart = dataBuffer.getBigInt64(colDataHead + 8, true);
+            const  decimalLowPart = dataBuffer.getBigUint64(colDataHead, true);
+            const decimalCombined = (decimalHighPart << 64n) | decimalLowPart;
+            return decimalToString(decimalCombined.toString(), fields_scale);
+        }
+        case TDengineTypeCode.DECIMAL64: {
+            let decimalVal = dataBuffer.getBigInt64(colDataHead, true);
+            return decimalToString(decimalVal.toString(), fields_scale);
+        }
         default: {
-            throw new WebSocketQueryInterFaceError(ErrorCode.ERR_UNSUPPORTED_TDENGINE_TYPE, `unspported type ${meta.type} for column ${meta.name}`)
+            throw new WebSocketQueryInterFaceError(ErrorCode.ERR_UNSUPPORTED_TDENGINE_TYPE, `unsupported type ${meta.type} for column ${meta.name}`)
         }
     }
 }
@@ -480,4 +530,12 @@ function bitPos(n:number):number {
 
 export function bitmapLen(n: number): number {
 	return ((n) + ((1 << 3) - 1)) >> 3
+}
+
+function getScaleFromRowBlock(buffer: DataView, colIndex: number, startOffset: number): number {
+    // for decimal: |___bytes___|__empty__|___prec___|__scale___|
+    let backupPos = buffer.byteOffset + startOffset + 28 + colIndex * 5 + 1;
+    let scaleBuffer = new DataView(buffer.buffer, backupPos);
+    let scale = scaleBuffer.getInt32(0, true);
+    return scale & 0xFF;
 }

nodejs/src/common/utils.ts

@@ -156,4 +156,24 @@ function comparePreReleases(pre1: string | null, pre2: string | null): number {
   return pre1.localeCompare(pre2);
 }
 
+export function decimalToString(valueStr: string, fields_scale: bigint | null): string {
+    let decimalStr = valueStr;
+    if (fields_scale && fields_scale > 0) {
+        const scale = Number(fields_scale);
+        const isNegative = decimalStr.startsWith('-');
+        const absStr = isNegative ? decimalStr.slice(1) : decimalStr;
+
+        if (absStr.length <= scale) {
+            // If the length of the number is less than or equal to the precision, add 0 before it.
+            const paddedStr = absStr.padStart(scale + 1, '0');
+            decimalStr = (isNegative ? '-' : '') + '0.' + paddedStr.slice(1);
+        } else {
+            // 在指定位置插入小数点
+            const integerPart = absStr.slice(0, absStr.length - scale);
+            const decimalPart = absStr.slice(absStr.length - scale);
+            decimalStr = (isNegative ? '-' : '') + integerPart + '.' + decimalPart;
+        }
+    }
+    return decimalStr;
+}
 

nodejs/src/stmt/wsProto.ts

@@ -11,7 +11,7 @@ export interface StmtMessageInfo {
 interface StmtParamsInfo {
     req_id: number;
     sql?: string | undefined | null;
-    stmt_id?: number | undefined | null;
+    stmt_id?: bigint | undefined | null;
     name?: string | undefined | null;
     tags?: Array<any> | undefined | null;
     paramArray?: Array<Array<any>> | undefined | null;
@@ -20,10 +20,10 @@ interface StmtParamsInfo {
 
 export class WsStmtQueryResponse extends WSQueryResponse {
     affected:number | undefined | null;
-    stmt_id?: number | undefined | null;
+    stmt_id?: bigint | undefined | null;
     constructor(resp:MessageResp) {
         super(resp);
-        this.stmt_id = resp.msg.stmt_id
+        this.stmt_id = BigInt(resp.msg.stmt_id)
         this.affected = resp.msg.affected
     }
 }
@@ -34,7 +34,7 @@ export const enum StmtBindType {
 }
 
 
-export function binaryBlockEncode(bindParams :StmtBindParams, bindType:StmtBindType, stmtId:number, reqId:bigint, row:number): ArrayBuffer {
+export function binaryBlockEncode(bindParams :StmtBindParams, bindType:StmtBindType, stmtId:bigint, reqId:bigint, row:number): ArrayBuffer {
     //Computing the length of data
     let columns = bindParams.getParams().length;
     let length = TDengineTypeLength['BIGINT'] * 4;
@@ -46,7 +46,7 @@ export function binaryBlockEncode(bindParams :StmtBindParams, bindType:StmtBindT
     let arrayView = new DataView(arrayBuffer)
 
     arrayView.setBigUint64(0, reqId, true);
-    arrayView.setBigUint64(8, BigInt(stmtId), true);
+    arrayView.setBigUint64(8, stmtId, true);
     arrayView.setBigUint64(16, BigInt(bindType), true);
     //version int32
     arrayView.setUint32(24, 1, true);

nodejs/src/stmt/wsStmt.ts

@@ -1,3 +1,4 @@
+import JSONBig from 'json-bigint';
 import { WsClient } from '../client/wsClient';
 import { ErrorCode, TDWebSocketClientError, TaosError, TaosResultError } from '../common/wsError';
 import { WsStmtQueryResponse, StmtMessageInfo, binaryBlockEncode, StmtBindType } from './wsProto';
@@ -8,7 +9,7 @@ import logger from '../common/log';
 
 export class WsStmt {
     private _wsClient: WsClient;
-    private _stmt_id: number | undefined | null;
+    private _stmt_id: bigint | undefined | null;
     private _precision:number = PrecisionLength['ms'];
 
     private lastAffected: number | undefined | null;
@@ -147,7 +148,7 @@ export class WsStmt {
         return await this.execute(queryMsg, false);
     }
 
-    public getStmtId(): number | undefined | null {
+    public getStmtId(): bigint | undefined | null {
         return this._stmt_id;
     } 
 
@@ -156,7 +157,7 @@ export class WsStmt {
             if (this._wsClient.getState() <= 0) {
                 throw new TDWebSocketClientError(ErrorCode.ERR_CONNECTION_CLOSED, "websocket connect has closed!");
             }
-            let reqMsg = JSON.stringify(queryMsg);
+            let reqMsg = JSONBig.stringify(queryMsg);
             if (register) {
                 let result = await this._wsClient.exec(reqMsg, false);
                 let resp = new WsStmtQueryResponse(result)

nodejs/src/tmq/constant.ts

@@ -102,27 +102,31 @@ export class TMQMessageType {
 }
 
 export class TMQBlockInfo {
-	rawBlock?: ArrayBuffer;
-	precision?: number;
-	schema:  Array<TMQRawDataSchema>;
-	tableName?: string;
+    rawBlock?: ArrayBuffer;
+    precision?: number;
+    schema:  Array<TMQRawDataSchema>;
+    tableName?: string;
     constructor() {
         this.schema = [];
     }
 }
 
 export class TMQRawDataSchema  {
-	colType: number;
-	flag:   number;
-	bytes:  bigint;
-	colID:  number
-	name:  string;
+    colType: number;
+    flag:   number;
+    bytes:  bigint;
+    colID:  number
+    name:  string;
+    precision: number;
+    scale: number;
     constructor() {
         this.bytes = BigInt(0);
         this.colID = -1;
         this.colType = -1;
         this.flag = -1;
         this.name = "";
+        this.scale = 0;
+        this.precision = 0;
 
     }
 }

nodejs/src/tmq/tmqResponse.ts

@@ -1,5 +1,5 @@
 import { WSQueryResponse } from "../client/wsResponse";
-import { ColumnsBlockType, TDengineTypeLength } from "../common/constant";
+import { ColumnsBlockType, TDengineTypeCode, TDengineTypeLength } from "../common/constant";
 import { MessageResp, TaosResult, _isVarType, getString, readBinary, readNchar, readSolidDataToArray, readVarchar } from "../common/taosResult";
 import { WebSocketInterfaceError, ErrorCode, TDWebSocketClientError } from "../common/wsError";
 import { TMQBlockInfo, TMQRawDataSchema } from "./constant";
@@ -15,7 +15,7 @@ export class WsPollResponse {
     topic: string;
     database: string;
     vgroup_id:number;
-    message_id: number;
+    message_id: bigint;
     id: bigint;
     message_type:number;
     totalTime:number;    
@@ -29,7 +29,7 @@ export class WsPollResponse {
         this.topic = resp.msg.topic;
         this.database = resp.msg.database;
         this.vgroup_id = resp.msg.vgroup_id;
-        this.message_id = resp.msg.message_id;
+        this.message_id = BigInt(resp.msg.message_id);
         this.message_type = resp.msg.message_type;
         if (resp.msg.id) {
             this.id = BigInt(resp.msg.id);
@@ -245,7 +245,7 @@ export class WSTmqFetchBlockInfo {
                     let bitMapArr = new Uint8Array(dataView.buffer, dataView.byteOffset + bufferOffset, bitMapOffset)
                     bufferOffset += bitMapOffset;
                     //decode column data, data is array
-                    data = readSolidDataToArray(dataView, bufferOffset, rows, this.schema[i].colType, bitMapArr);
+                    data = readSolidDataToArray(dataView, bufferOffset, rows, this.schema[i].colType, bitMapArr, startOffset, i);
 
                 } else {  
                     //Variable length type   
@@ -347,15 +347,15 @@ export class PartitionsResp{
     action: string;
     totalTime: number;
     timing:bigint;
-    positions:number[];
+    positions:bigint[];
     constructor(resp:MessageResp) {
         this.timing = BigInt(resp.msg.timing);
         this.code = resp.msg.code;
         this.message = resp.msg.message;
         this.req_id = resp.msg.req_id;
         this.action = resp.msg.action;
         this.totalTime = resp.totalTime;
-        this.positions = resp.msg.position;
+        this.positions = resp.msg.position ? resp.msg.position.map((pos: number) => BigInt(pos)) : [];
     }
 
     setTopicPartitions(topicPartitions:TopicPartition[]):TopicPartition[] {
@@ -373,21 +373,21 @@ export class PartitionsResp{
 export class CommittedResp extends PartitionsResp {
     constructor(resp:MessageResp) {
         super(resp);
-        this.positions = resp.msg.committed
+        this.positions = resp.msg.committed ? resp.msg.committed.map((pos: number) => BigInt(pos)) : [];
     }
 }
 
 export class TopicPartition {
     topic       :string;
     vgroup_id   :number;
-    offset      ?:number;
-    begin       ?:number;
-    end         ?:number;
+    offset      ?:bigint;
+    begin       ?:bigint;
+    end         ?:bigint;
     constructor(msg:any) {
         this.vgroup_id = msg.vgroup_id;
-        this.offset = msg.offset;
-        this.begin = msg.begin;
-        this.end = msg.end;
+        this.offset = BigInt(msg.offset);
+        this.begin = BigInt(msg.begin);
+        this.end = BigInt(msg.end);
         this.topic = ''
     }
 }