feat: respect iox::column_type::field metadata when mapping query (#491)

CHANGELOG.md

@@ -1,3 +1,15 @@
+## 1.0.0 [unreleased]
+
+### Features
+
+1. [#491](https://github.com/InfluxCommunity/influxdb3-js/pull/491): Respect iox::column_type::field metadata when
+   mapping query results into values.
+    - iox::column_type::field::integer: => number
+    - iox::column_type::field::uinteger: => number
+    - iox::column_type::field::float: => number
+    - iox::column_type::field::string: => string
+    - iox::column_type::field::boolean: => boolean
+
 ## 0.13.0 [unreleased]
 
 ### Features

packages/client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@influxdata/influxdb3-client",
-  "version": "0.12.0",
+  "version": "1.0.0",
   "description": "The Client that provides a simple and convenient way to interact with InfluxDB 3.",
   "scripts": {
     "build": "yarn cp ../../README.md ./README.md && yarn run clean && yarn run build:browser && yarn run build:node",

packages/client/src/impl/QueryApiImpl.ts

@@ -9,6 +9,7 @@ import {impl} from './implSelector'
 import {PointFieldType, PointValues} from '../PointValues'
 import {allParamsMatched, queryHasParams} from '../util/sql'
 import {CLIENT_LIB_USER_AGENT} from './version'
+import {getMappedValue} from '../util/TypeCasting'
 
 export type TicketDataType = {
   database: string
@@ -117,7 +118,8 @@ export default class QueryApiImpl implements QueryApi {
       const row: Record<string, any> = {}
       for (const batchRow of batch) {
         for (const column of batch.schema.fields) {
-          row[column.name] = batchRow[column.name]
+          const value = batchRow[column.name]
+          row[column.name] = getMappedValue(column, value)
         }
         yield row
       }
@@ -164,8 +166,10 @@ export default class QueryApiImpl implements QueryApi {
           const [, , valueType, _fieldType] = metaType.split('::')
 
           if (valueType === 'field') {
-            if (_fieldType && value !== undefined && value !== null)
-              values.setField(name, value, _fieldType as PointFieldType)
+            if (_fieldType && value !== undefined && value !== null) {
+              const mappedValue = getMappedValue(columnSchema, value)
+              values.setField(name, mappedValue, _fieldType as PointFieldType)
+            }
           } else if (valueType === 'tag') {
             values.setTag(name, value)
           } else if (valueType === 'timestamp') {

packages/client/src/impl/version.ts

@@ -1,2 +1,2 @@
-export const CLIENT_LIB_VERSION = '0.12.0'
+export const CLIENT_LIB_VERSION = '1.0.0'
 export const CLIENT_LIB_USER_AGENT = `influxdb3-js/${CLIENT_LIB_VERSION}`

packages/client/src/index.ts

@@ -6,7 +6,7 @@ export * from './util/logger'
 export * from './util/escape'
 export * from './util/time'
 export * from './util/generics'
-export {collectAll} from './util/common'
+export {collectAll, isNumber} from './util/common'
 export * from './Point'
 export * from './PointValues'
 export {default as InfluxDBClient} from './InfluxDBClient'

packages/client/src/util/TypeCasting.ts

@@ -0,0 +1,63 @@
+import {Field} from 'apache-arrow'
+import {isNumber, isUnsignedNumber} from './common'
+import {Type as ArrowType} from 'apache-arrow/enum'
+
+/**
+ * Function to cast value return base on metadata from InfluxDB.
+ *
+ * @param field the Field object from Arrow
+ * @param value the value to cast
+ * @return the value with the correct type
+ */
+export function getMappedValue(field: Field, value: any): any {
+  if (value === null || value === undefined) {
+    return null
+  }
+
+  const metaType = field.metadata.get('iox::column::type')
+
+  if (!metaType || field.typeId === ArrowType.Timestamp) {
+    return value
+  }
+
+  const [, , valueType, _fieldType] = metaType.split('::')
+
+  if (valueType === 'field') {
+    switch (_fieldType) {
+      case 'integer':
+        if (isNumber(value)) {
+          return parseInt(value)
+        }
+        console.warn(`Value ${value} is not an integer`)
+        return value
+      case 'uinteger':
+        if (isUnsignedNumber(value)) {
+          return parseInt(value)
+        }
+        console.warn(`Value ${value} is not an unsigned integer`)
+        return value
+      case 'float':
+        if (isNumber(value)) {
+          return parseFloat(value)
+        }
+        console.warn(`Value ${value} is not a float`)
+        return value
+      case 'boolean':
+        if (typeof value === 'boolean') {
+          return value
+        }
+        console.warn(`Value ${value} is not a boolean`)
+        return value
+      case 'string':
+        if (typeof value === 'string') {
+          return String(value)
+        }
+        console.warn(`Value ${value} is not a string`)
+        return value
+      default:
+        return value
+    }
+  }
+
+  return value
+}

packages/client/src/util/common.ts

@@ -35,3 +35,42 @@ export const collectAll = async <T>(
   }
   return results
 }
+
+/**
+ * Check if an input value is a valid number.
+ *
+ * @param value - The value to check
+ * @returns Returns true if the value is a valid number else false
+ */
+export const isNumber = (value?: number | string | null): boolean => {
+  if (value === null || undefined) {
+    return false
+  }
+
+  if (
+    typeof value === 'string' &&
+    (value === '' || value.indexOf(' ') !== -1)
+  ) {
+    return false
+  }
+
+  return value !== '' && !isNaN(Number(value?.toString()))
+}
+
+/**
+ * Check if an input value is a valid unsigned number.
+ *
+ * @param value - The value to check
+ * @returns Returns true if the value is a valid unsigned number else false
+ */
+export const isUnsignedNumber = (value?: number | string | null): boolean => {
+  if (!isNumber(value)) {
+    return false
+  }
+
+  if (typeof value === 'string') {
+    return Number(value) >= 0
+  }
+
+  return typeof value === 'number' && value >= 0
+}

packages/client/test/integration/e2e.test.ts

@@ -1,7 +1,6 @@
 import {expect} from 'chai'
-import {InfluxDBClient, Point} from '../../src'
+import {InfluxDBClient, Point, PointValues} from '../../src'
 import {rejects} from 'assert'
-import {PointValues} from '../../src'
 ;(BigInt.prototype as any).toJSON = function () {
   return this.toString()
 }
@@ -68,7 +67,9 @@ describe('e2e test', () => {
       .setIntegerField('testId', testId)
     await client.write(point, database)
 
-    const query = `SELECT * FROM "stat" WHERE "testId" = ${testId}`
+    const query = `SELECT *
+                   FROM "stat"
+                   WHERE "testId" = ${testId}`
 
     const data = client.query(query, database)
 
@@ -99,7 +100,7 @@ describe('e2e test', () => {
     // test aggregation query
     //
     const queryAggregation = `
-    SELECT sum("avg") as "sum_avg", sum("max") as "sum_max"
+        SELECT sum("avg") as "sum_avg", sum("max") as "sum_max"
         FROM "stat"
         WHERE "testId" = ${testId}
     `
@@ -147,12 +148,12 @@ describe('e2e test', () => {
     await sleep(2_000)
 
     const query = `
-      SELECT *
+        SELECT *
         FROM "stat"
         WHERE
-        time >= now() - interval '10 minute'
-        AND
-        "testId" = ${testId}
+            time >= now() - interval '10 minute'
+          AND
+            "testId" = ${testId}
     `
     const paralelQueries = 8
 
@@ -224,12 +225,12 @@ describe('e2e test', () => {
     await sleep(5_000)
 
     const query = `
-      SELECT *
+        SELECT *
         FROM "stat"
         WHERE
-        time >= now() - interval '10 minute'
-        AND
-        "testId" = ${testId}
+            time >= now() - interval '10 minute'
+          AND
+            "testId" = ${testId}
     `
     const queryValues: typeof values = []
 
@@ -258,6 +259,49 @@ describe('e2e test', () => {
     await client.close()
   }).timeout(40_000)
 
+  it('queryPoints with getMappedValue', async () => {
+    const {database, token, url} = getEnvVariables()
+
+    const client = new InfluxDBClient({
+      host: url,
+      token,
+      writeOptions: {
+        precision: 'ms',
+      },
+    })
+
+    const time = Date.now()
+    const testId = getRandomInt(0, 100000000)
+    await client.write(
+      `host15,tag=empty name="intel",mem_total=2048,disk_free=100i,temperature=100.86,isActive=true,testId="${testId}" ${time}`,
+      database
+    )
+
+    const sql = `Select *
+                 from host15
+                 where "testId" = ${testId}`
+    const dataPoints = client.queryPoints(sql, database)
+
+    const pointRow: IteratorResult<PointValues, void> = await dataPoints.next()
+
+    expect(pointRow.value?.getField('name')).to.equal('intel')
+    expect(pointRow.value?.getFieldType('name')).to.equal('string')
+
+    expect(pointRow.value?.getField('mem_total')).to.equal(2048)
+    expect(pointRow.value?.getFieldType('mem_total')).to.equal('float')
+
+    expect(pointRow.value?.getField('disk_free')).to.equal(100)
+    expect(pointRow.value?.getFieldType('disk_free')).to.equal('integer')
+
+    expect(pointRow.value?.getField('temperature')).to.equal(100.86)
+    expect(pointRow.value?.getFieldType('temperature')).to.equal('float')
+
+    expect(pointRow.value?.getField('isActive')).to.equal(true)
+    expect(pointRow.value?.getFieldType('isActive')).to.equal('boolean')
+
+    await client.close()
+  }).timeout(10_000)
+
   const samples = [
     {
       measurement: 'frame',
@@ -310,6 +354,7 @@ describe('e2e test', () => {
       quality: 'Excellent',
     },
   ]
+
   async function writeFrameSamples(client: InfluxDBClient, database: string) {
     const time = Date.now()
 
@@ -346,12 +391,12 @@ describe('e2e test', () => {
     await sleep(3_000)
 
     const query = `
-      SELECT *
+        SELECT *
         FROM "${samples[0].measurement}"
         WHERE
-        time >= now() - interval '10 minute'
-        AND
-        "director" = $director    
+            time >= now() - interval '10 minute'
+          AND
+            "director" = $director
     `
     const data = client.query(query, database, {
       type: 'sql',
@@ -364,7 +409,7 @@ describe('e2e test', () => {
       expect(row['director']).to.equal('J_Ford')
     }
     expect(count).to.be.greaterThan(0)
-  }).timeout(5_000)
+  }).timeout(10_000)
 
   it('queries to points with parameters', async () => {
     const {database, token, url} = getEnvVariables()
@@ -382,12 +427,12 @@ describe('e2e test', () => {
     await sleep(3_000)
 
     const query = `
-      SELECT *
+        SELECT *
         FROM "${samples[0].measurement}"
         WHERE
-        time >= now() - interval '10 minute'
-        AND
-        "director" = $director    
+            time >= now() - interval '10 minute'
+          AND
+            "director" = $director
     `
     const points = client.queryPoints(query, database, {
       type: 'sql',
@@ -419,11 +464,12 @@ describe('e2e test', () => {
     await sleep(3_000)
 
     const query = `SELECT *
-FROM "frame"
-WHERE
-time > now() - 1h
-AND
-"director" = 'H_Hathaway'`
+                   FROM "frame"
+                   WHERE
+                       time
+                       > now() - 1h
+                     AND
+                       "director" = 'H_Hathaway'`
 
     const points = client.queryPoints(query, database)