Transform a resultSet into a readable stream

[jgoux]

Hello,
I want to turn a resultSet into a readableStream (like node-mssql and node-mysql provide through their pipe methods), so I can pipe it to whatever I want to do with the results (millions of lines).
Here is my function to execute a query and then returns a readable stream :

async function executeQuery(connectionParameters, query, queryParameters) {
    const getConnection = promisify(oracle.getConnection.bind(oracle))
    const connection = await getConnection(connectionParameters)

    const executeQuery = promisify(connection.execute.bind(connection))
    const result = await executeQuery(query, queryParameters, { resultSet: true })

    const getRow = promisify(result.resultSet.getRow.bind(result.resultSet))

    const rs = Readable({ objectMode: true })
    rs._read = async function() {
        const row = await getRow()
        if (!row) {
            result.resultSet.close((e) => { if (e) throw e })
            connection.release((e) => { if (e) throw e })
            return rs.push(null)
        }
        rs.push(row)
    }
    return rs
}

Here is how I use it :

const query$ = await executeQuery(connectionParameters, query, queryParameters)

    return new Promise((resolve, reject) => {
        query$.on("error", reject)

        const csvStringifier$ = csvStringify({ delimiter: argv.delimiter })
        csvStringifier$.on("error", reject)

        const file$ = fs.createWriteStream("./extraction.csv")
        file$.on("finish", resolve).on("error", reject)

        query$.pipe(csvStringifier$).pipe(file$)
    })

My issue is as a certain point, the fetching stops without errors.
For the same query, sometimes they're 10K rows written to the file, then only 2K, then 14K...
I'm pretty sure it's due to my _read implementation, but I can't figure out what's wrong with it.
In node.js doc (https://nodejs.org/api/stream.html#stream_readable_read_size_1) this part is unclear to me :

When _read is called, if data is available from the resource, _read should start pushing that data into the read queue by calling this.push(dataChunk). _read should continue reading from the resource and pushing data until push returns false, at which point it should stop reading from the resource. Only when _read is called again after it has stopped should it start reading more data from the resource and pushing that data onto the queue.

Thanks in advance.

EDIT : Here is an example of an implementation : https://github.com/felixge/node-mysql/blob/de5913227dbbaacbbdf22ee38689d090e1451be9/lib/protocol/sequences/Query.js#L188

[cjbj]

@jgoux did you get any further with this?

[leguma]

From the documentation you quoted, I believe it wants you to keep pushing data until you either reach the end or it tells you to stop (when push returns false). That way, the other end of the pipe will keep getting data until they pause it (or it ends).

    rs._read = async function() {
        do {
            row = await getRow()
            if (!row) {
                result.resultSet.close((e) => { if (e) throw e })
                connection.release((e) => { if (e) throw e })
                return rs.push(null)
            }
        } while (rs.push(row))
    }

[leguma]

Even with the corrected stream interface, I have problems. For me, all data pushed doesn't seem to actually go through the pipe, and the stream never emits its end/close events. Also, piping the stream directly only results in one record being fetched, while wrapping the pipe in a Promise (like your example usage) somehow returns more. Very odd. I strongly suspect it has something to do with transpiling async/awaits.

I started implementing it in non-transpiled code, but quickly understood why you took the route you did: getRow doesn't actually return the data, it only provides the callback that you need to get data from an unrelated event (_read). Ugh... We could actually probably implement it pretty easily if getRow was synchronous, funny enough.

@cjbj, any chance we can avoid callback hell and have a nice, pretty Stream interface like blobs/clobs? Bonus points if you can eliminate some backend traffic by grabbing whatever rows are "ready" on each DB call (up to a certain limit to prevent memory bloat, of course).

EDIT:
Actually, I figured out my issue. Somewhere along the line, I forgot to change my outFormat, so I was expecting objects and getting arrays, which was causing silent errors in my streams. Whoops.

async function executeQuery(connectionParameters, query, queryParameters) {
    const getConnection = promisify(oracle.getConnection.bind(oracle));
    const connection = await getConnection(connectionParameters);

    const exQuery = promisify(connection.execute.bind(connection));
    const result = await exQuery(query, queryParameters, { resultSet: true, outFormat: oracle.OBJECT });
    const getRow = promisify(result.resultSet.getRow.bind(result.resultSet));

    const rs = Readable({ objectMode: true });
    rs._read = async function(size) {
        var row = null;
        do {
            row = await getRow();
            if (!row) {
                result.resultSet.close((e) => { if (e) throw e });
                connection.release((e) => { if (e) throw e });
                return rs.push(null);
            }
        } while (rs.push(row));
    }

    return rs;
}

Might want to pass down that as a parameter or something. Or, if you don't want to wrestle with transpiling:

util.inherits(OracleStream, Readable);

function OracleStream(connectionParameters, query, queryParameters, opt) {
    opt.objectMode = true;
    opt.resultSet = true;
    Readable.call(this, opt);

    this._isReading = true;
    this._rowsPerFetch = opt.rowsPerFetch || 1;

    oracle.getConnection(connectionParameters, function(err, connection) {
        if (err) { console.error(err.message); return; }

        this._connection = connection;
        this._connection.execute(query, queryParameters, opt, function(err, result) {
            if (err) {
                console.error(err.message);
                this._connection.release((e) => { if (e) throw e });
                return;
            }

            this._resultSet = result.resultSet;
            this.fetchRows();
        }.bind(this));
    }.bind(this));
}

OracleStream.prototype._read = function(size) {
    if (!this._isReading) {
        this._isReading = true;
        this.fetchRows();
    }
}

OracleStream.prototype.fetchRows = function() {
    this._resultSet.getRows(this._rowsPerFetch, function (err, rows) {
        if (err || !rows || rows.length == 0) {
            if (err) {
                console.error(err.message);
            }

            this._resultSet.close(function(err) { 
                if (err) throw err;
                this._connection.release((e) => { if (e) throw e });
            }.bind(this));

            this.push(null);
            this._isReading = false;
            return;
        }

        var lowWater = true;
        for (var i = 0; i < rows.length; i++) {
            lowWater = this.push(rows[i]);
        }

        if (lowWater) {
            this.fetchRows();
        } else {
            this._isReading = false;
        }
    }.bind(this));
}

exports.createReadStream = function(connectionParameters, query, queryParameters, opt) {
    return new OracleStream(connectionParameters, query, queryParameters, opt);
}

// Example Usage
const oracleStream = require('./oracleStream.js');

const query$ = oracleStream.createReadStream(dbConfig, query, queryParameters, { outFormat: oracle.OBJECT, rowsPerFetch: 2 });
const json$ = JSONStream.stringify();
const file$ = fs.createWriteStream("./example.json");

query$.pipe(json$).pipe(file$);

[leguma]

Based on other _read implementations I've seen, it appears that the most common strategy is to only push one "chunk" until the next read is called (like the OP's implementation). However, in my experimentation I only saw _read called once at the start of the pipe for tiny streams (since the buffers didn't fill up), which supports the idea that you should keep filling up the pipe until you reach the high water mark.

For large amounts of data (that reach the high water mark during the stream), _read was called several times sequentially to resume the flow (think of an impatient user clicking a button a bunch of times). In my initial implementation, I forgot the this._isReading = true; line in _read, which caused fetchRows to be called several times in parallel as a result. A limitation of oracledb is that the resultSet can only be accessed sequentially; attempting to get data in parallel throws an error indicating such.

In summary, I believe the OP's error was indeed in the _read implementation, specifically that there needed to be preventive measures taken to ensure that the _read doesn't fire parallel getRow calls. My second implementation (the non-transpiled version) handles this issue by preventing parallel calls. If Node.js was multi-threaded, this would have the race condition on the running check, but you should be safe since it handles thing with a single thread (and children via clusters would each have their own request they're handling).

Either use my second solution, or add an _isReading check to the transpiled solution and I suspect your problem should go away.

[cjbj]

@leguma many thanks. I need to test it out.

Regarding your question in #219 (comment) , I know we have other things on the list before we revisit the node-oracledb API. The network optimization already exists in the form of pre-fetching

[jgoux]

Thanks @leguma for your time invested in this issue.
I'll try your implementation and give you a feedback. 👍

[sagiegurari]

I now had the same requirement to handle huge amounts of results so I modified the oracledb wrapper with ability to stream bulks of rows (currently 100 rows at a time, but can be changed via bulkRowsAmount option).
https://github.com/sagiegurari/simple-oracledb#usage-query

connection.query('SELECT * FROM departments WHERE manager_id > :id', [110], {
  streamResults: true,
  bulkRowsAmount: 100 //The amount of rows to fetch (for streaming, thats the max rows that the callback will get for each streaming invocation)
}, function onResults(error, results) {
  if (error) {
    //handle error...
  } else if (results.length) {
    //handle next bulk of results
  } else {
    //all rows read
  }
});

[sagiegurari]

updated the example above

[leguma]

@sagiegurari
Note that ResultSets already have a rowsPerFetch option and a callback function. Unless there's some sort of lingering overhead with ResultSets (such as the full results being stored somewhere), then I don't think this is any different (and could be worse). What's the difference? Are you making several execute calls with a ROWNUM param or something?

Also, the desire was to use the Node.js streams interface, so we can do things like pipe these into other streams. For an example, see the original post.

[sagiegurari]

@leguma the difference is that with resultset you need to do getRows over and over until you finish.
thats a lot of messy code that you need to write everywhere.
So, I actually had a need for getting bulks over and over in same callback until done (without a need to write lots of coding to fetch next rows and so on with result set) which really simplifies things.
than on top of those bulks, in our app code (not in my lib) i wrote a stream api which gave bulks of rows via stream events.
I could move to my library actually, not sure why i didn't put it there originally.

[sagiegurari]

updated the streamResults to splitResults to be more clear based on your feedback and i'll probably add the streaming that i wrote in the lib later on to enabling piping.

[leguma]

https://github.com/oracle/node-oracledb/blob/master/doc/api.md#-912-result-set-handling

While it is a bit nicer to have the callback attached directly, it doesn't really take that many more lines of code with ResultSets (I'm counting ~10). Also, you don't have to call getRows everywhere; in the documentation, a recursive call is given as an example, which basically works like a callback in this case. Giving us the opportunity to call getRows manually allows us to be flexible and throttle/pause reading data if necessary, which is critically important for implementing the Node.js stream interface. If I were to try to implement the same with a mandatory onResults callback, then we'd be forced to store all the data in the event that the Node.js streams' buffers get filled and we have to pause writing to them. That means we'd run out of memory in a hurry.

I'm interested to see the stream implementation you've mentioned adding. Perhaps you've resolved these issues there.

As a side-note, I did some profiling with the fetch size for resultSets. It seems that the optimal fetch size differs based on:

• The expected number of items retrieved.
• Your machine's processing power (ie how fast it can process the series of piped streams).
• The complexity of the query being called.

Some preliminary (by no means exhaustive) testing, it seems like a fetch size of 2-3 works best for my streaming implementation for a simple query. When I have some free time, I might compile a performance analysis with different query complexities and fetch sizes.

[sagiegurari]

The thing you are forgetting, is that the example you gave doesn't count for LOB handling which adds more async code.
take that + the fact that you do it in many places and it turns into a real problem to maintain.
The code i give already handles LOBs and transforms them into string or buffers (depending if clob or blob).
So you get a nice js object with fields.

the stream implementation i had was pretty basic and I worked with events and not pipe, so it is not tested for that.
I'll refactor it to decouple from our app and run few tests to see if it is good enough to incorporate in the simple-oracledb.
For me it was a fairly new use case to pull huge amounts of rows and stream them out, as I mostly worked with few lines of json data stored in CLOBs which i need to query/insert/update, so I only very recently developed the bulks of data to stream tens of thousands of rows (no.... not millions thank god. I still use nodejs mainly as a web server).

[leguma]

Right you are: my implementation even forces the response to be returned as objects, which may be less than optimal (probably resulting in a significant overhead for large amounts of data). This was mostly for convenience in passing the data to a jsonify stream. It would probably be more efficient to use the non-object version of execute and using the header row as a json template, then mapping each record into it. I digress.

Luckily, LOBs are already streams in node-oracledb. Haven't tried 'em myself, but the syntax seems pretty easy for reading them. Writing them does look quite verbose, with non-intuitive configuration requirements (autoCommit=false and bound out vars), but I think it was done that way to have a flexible, generic execute function. A prettified interface with most of that stuff preconfigured is quite desirable. Something like

// Get CLOB read stream
const clobStream$ = oracleStream.createLOBReadStream(dbConfig, query, queryParameters, { type=oracledb.CLOB });

// Get BLOB write stream
const blobWriteStream$ = oracleStream.createLOBWriteStream(dbConfig, query, queryParameters, { type=oracledb.BLOB });

Also, would be cool to get a streaming write interface set up for non-LOBs. Even more interesting might be making a duplex stream where you can pipe read/write data in and out. Example use case: need to run selects on a large set of data, so you pipe the query data in, it runs the query on each item (or chunk of items), and pipes the data out. However, I don't really anticipate that as a commonly needed operation. More of a fun project.