Add indbstream option to enable record-by-record data processing

Author: nmaguiar

data/completion.yaml

@@ -32,6 +32,8 @@ complete:
       desc: The JDBC access timeout
     - name: indblib=
       desc: Use a JDBC driver oPack generated by ojob.io/db/getDriver
+    - name: indbstream=
+      desc: If true will stream the data instead of loading it all in memory
     - name: indbexec=
       desc: If true the input SQL is not a query but a DML statement
   - name: in=gb64json

data/usage.json

@@ -694,6 +694,11 @@
       "Type": "String",
       "Description": "Use a JDBC driver oPack generated by ojob.io/db/getDriver"
     },
+    {
+      "Option": "indbstream",
+      "Type": "Boolean",
+      "Description": "If true the output will be processed record by record"
+    },
     {
       "Option": "indbexec",
       "Type": "Boolean",

src/docs/USAGE.md

@@ -240,6 +240,7 @@ List of options to use when _in=db_ (SQL query):
 | indbpass | String | The JDBC access password |
 | indbtimeout | String | The JDBC access timeout |
 | indblib | String | Use a JDBC driver oPack generated by ojob.io/db/getDriver |
+| indbstream | Boolean | If true the output will be processed record by record |
 | indbexec | Boolean | If true the input SQL is not a query but a DML statement | 
 
 > JDBC oracle: jdbc:oracle:thin:@[host]:[port]:[database]

src/include/inputFns.js

@@ -411,11 +411,58 @@ var _inputFns = new Map([
                     _$o({ affectedRows: _r }, options)
                     _db.commit()
                 } else {
-                    var _r = _db.q(r)
-                    if (isMap(_r) && isArray(_r.results)) {
-                        _$o(_r.results, options)
-                    } else {
-                        _exit(-1, "Invalid DB result: " + stringify(_r))
+                    if (toBoolean(params.indbstream)) {
+                        var _rs = _db.qsRS(r)
+                        try {
+                            while(_rs.next()) {
+                                var _r = {}
+                                for (var i = 1; i <= _rs.getMetaData().getColumnCount(); i++) {
+                                    var _v = _rs.getObject(i)
+                                    switch(_rs.getMetaData().getColumnType(i)) {
+                                    case java.sql.Types.BIGINT:
+                                    case java.sql.Types.INTEGER:
+                                    case java.sql.Types.TINYINT:
+                                    case java.sql.Types.SMALLINT:
+                                    case java.sql.Types.NUMERIC:
+                                        _v = Number(_v)
+                                        break
+                                    case java.sql.Types.DOUBLE:
+                                    case java.sql.Types.FLOAT:
+                                    case java.sql.Types.REAL:
+                                    case java.sql.Types.DECIMAL:
+                                        _v = Number(_v)
+                                        break
+                                    case java.sql.Types.BOOLEAN:
+                                        _v = Boolean(_v)
+                                        break
+                                    case java.sql.Types.TIME:
+                                    case java.sql.Types.DATE:
+                                    case java.sql.Types.TIMESTAMP:
+                                        _v = new Date(_v.getTime())
+                                        break
+                                    case java.sql.Types.NULL:
+                                        _v = null
+                                        break
+                                    default:
+                                        _v = String(_v)
+                                    }
+                                    _r[_rs.getMetaData().getColumnName(i)] = _v
+                                }
+                                _$o(_r, options)
+                            }
+                        } catch(e) {
+                            _exit(-1, "Error streaming SQL: " + e.message)
+                        } finally {
+                            _db.closeStatement(r)
+                            _rs.close()
+                        }
+                    } else {
+                        var _r = _db.q(r)
+                        if (isMap(_r) && isArray(_r.results)) {
+                            _$o(_r.results, options)
+                        } else {
+                            _exit(-1, "Invalid DB result: " + stringify(_r))
+                        }
                     }
                 }
             } catch(edb) {

src/include/transformFns.js

@@ -562,7 +562,7 @@ var _transformFns = {
         let _lst = params.field2date.split(",").map(r => r.trim())
         traverse(_r, (aK, aV, aP, aO) => {
             if (_lst.indexOf(aP.length > 0 && !aP.startsWith("[") ? aP.substring(1) + "." + aK : aK) >= 0 && isNumber(aV) && aV > 0) {
-                try { aO[aK] = ow.format.fromISODate(aV) } catch(e) {}
+                try { aO[aK] = isString(aV) ? ow.format.fromISODate(aV) : new Date(aV) } catch(e) { printErr(e) }
             }
         })
         return _r