Timestamp to UTC (#64)

* coeercing timestamp to UTC

* callback

Author: Sylvestre

lib/databricks.js

@@ -1,5 +1,5 @@
 import {json} from "micro";
-import {Readable} from "node:stream";
+import {Readable, Transform} from "node:stream";
 import JSONStream from "JSONStream";
 
 import {
@@ -226,13 +226,19 @@ export async function queryStream(req, res, connection) {
   try {
     query = await session.executeStatement(sql, {runAsync: true});
     const rows = await query.fetchAll();
+    const timestampFieldMap = new Map();
     const schema = await query.getSchema();
 
     const responseSchema = {
       type: "array",
       items: {
         type: "object",
-        properties: schema.columns.reduce((schema, col, idx) => {
+        properties: schema.columns.reduce((schema, col) => {
+          // If the column is a timestamp, we need keep track of its key.
+          // We will use this later to convert the timestamp to a UTC ISO-96801 format.
+          if (col.typeDesc.types[0].primitiveEntry.type === 8) {
+            timestampFieldMap.set(col.columnName, "TIMESTAMP");
+          }
           return {
             ...schema,
             ...{
@@ -261,7 +267,32 @@ export async function queryStream(req, res, connection) {
 
       stream.on("error", reject);
 
-      stream.pipe(JSONStream.stringify("", "\n", "\n")).pipe(res);
+      stream
+        .pipe(
+          new Transform({
+            objectMode: true,
+            transform(chunk, encoding, callback) {
+              let row = null;
+              try {
+                row = Object.entries(chunk).reduce((row, [key, value]) => {
+                  return {
+                    ...row,
+                    ...{
+                      [key]: timestampFieldMap.has(key) ? `${value}Z` : value,
+                    },
+                  };
+                }, {});
+              } catch (e) {
+                console.error("row has unexpected format");
+                // TODO: Add error handling once server supports handling error for in flight streamed response
+                // cb(new Error(e));
+              }
+              callback(null, row);
+            },
+          })
+        )
+        .pipe(JSONStream.stringify("", "\n", "\n"))
+        .pipe(res);
     });
   } catch (error) {
     if (!error.statusCode) error.statusCode = 400;