Help with queryStream

[ChrisHAdams]

I've been using node-oracledb for a few months and I've managed to achieve what I have needed to so far.

I'm currently working on a search app that could potentially return about 2m rows of data from a single call. To ensure I don't get a disconnect from the browser and the server, I thought I would try queryStream so that there is a constant flow of data back to the client.

I implemented the queryStream example as-is, and this worked fine for a few hundred thousand rows. However, when the returned rows is greater than one million, Node runs out of memory. By logging and watching both client and server log events, I can see that client is way behind the server in terms of rows sent and received.

It's worth noting that at this point, my selectstream implementation is within a req/res function called via Express.

To return the data, I do something like....

    stream.on('data', function (data) {
        // console.log("stream 'data' event");
        // console.log(data);
        rowcount++;

        let obj = new myObjectConstructor(data);
        res.write(JSON.stringify(obj.getJson());

    });

I've been reading about how streams and pipe can help with flow, so what I'd like to be able to do is to be able to pipe the results from the query to a) help with flow and b) to be able to pipe the results to other functions before sending back to the client.

E.g.

function getData(req, res){

    var stream = myQueryStream(connection, query);

    stream
        .pipe(toSomeOtherFunction)
        .pipe(yetAnotherFunction)
        .pipe(res);

}

I'm spent a few hours trying to find a solution or example that allows me to pipe results, but I'm stuck and need some help.

Apologies if I'm missing something obvious, but I'm still getting to grips with Node and especially streams.

Thanks in advance.

[cjbj]

@dmcghan any thoughts?

[dmcghan]

@ChrisHAdams Sorry for the delay, lots of travel lately...

There's a bit of an impedance mismatch here. The queryStream API emits rows of JavaScript objects, but what you want to stream to the client is a JSON array. You basically have to add an open bracket to the beginning, a comma after each row, and a close bracket to the end.

I'll show you how to do this in a controller that uses the driver directly as you have done, instead of using separate database modules as I advocate in this series.

const oracledb = require('oracledb');

async function get(req, res, next) {
  try {
    const conn = await oracledb.getConnection();

    const stream = await conn.queryStream('select * from employees', [], {outFormat: oracledb.OBJECT});

    res.writeHead(200, {'Content-Type': 'application/json'});

    res.write('[');

    stream.on('data', (row) => {
      res.write(JSON.stringify(row));
      res.write(',');
    });

    stream.on('end', () => {
      res.end(']');
    });

    stream.on('close', async () => {
      try {
        await conn.close();
      } catch (err) {
        console.log(err);
      }
    });

    stream.on('error', async (err) => {
      next(err);
      
      try {
        await conn.close();
      } catch (err) {
        console.log(err);
      }
    });
  } catch (err) {
    next(err);
  }
}

module.exports.get = get;

Once you get the concepts, you can simplify things a bit with a reusable Transform class which allows you to use pipe in the controller logic:

const oracledb = require('oracledb');
const { Transform } = require('stream');

class ToJSONArray extends Transform {
  constructor() {
    super({objectMode: true});

    this.push('[');
  }

  _transform (row, encoding, callback) {
    if (this._prevRow) {
      this.push(JSON.stringify(this._prevRow));
      this.push(',');
    }

    this._prevRow = row;

    callback(null);
  }

  _flush (done) {
    if (this._prevRow) {
      this.push(JSON.stringify(this._prevRow));
    }

    this.push(']');

    delete this._prevRow;

    done();
  }
}

async function get(req, res, next) {
  try {
    const toJSONArray = new ToJSONArray();
    const conn = await oracledb.getConnection();

    const stream = await conn.queryStream('select * from employees', [], {outFormat: oracledb.OBJECT});

    res.writeHead(200, {'Content-Type': 'application/json'});

    stream.pipe(toJSONArray).pipe(res);

    stream.on('close', async () => {
      try {
        await conn.close();
      } catch (err) {
        console.log(err);
      }
    });

    stream.on('error', async (err) => {
      next(err);

      try {
        await conn.close();
      } catch (err) {
        console.log(err);
      }
    });
  } catch (err) {
    next(err);
  }
}

module.exports.get = get;

Having said all of that, I think we may be able to add something to core to simplify this some. I'll float some ideas by the team to see what they think...

[ChrisHAdams]

Hi Dan,

That's brilliant. Many, many thanks for taking the time to put together such a comprehensive reply.

The bit I was struggling with was moving from the stream.on('some event') approach to being able to pipe the results from one function to another in the controller, without cluttering the controller with the DAO parts.

All other parts of my app have controller layer and a data access layer, but for this I had them together just whilst I was quickly trying stuff out, but with your example I can now see how I can implement in the way I want to and how to separate the code into logical layers.

I'll look forward to using the above in my app.

Oh, thank you for the link too. I'll read in more depth tomorrow.

Regards,

Chris

[bchr02]

@dmcghan
Thanks for this code example. To save the next reader some troubleshooting, you may want to update your code here:

stream.on('data', (row) => {
      res.write(JSON.stringify(row));
      res.write(',');
    });

to

let rowcount = 0;
stream.on('data', (row) => {
      if (rowcount++ != 0) res.write(",");
      res.write(JSON.stringify(row));
    });

The way you have it will always end the array within the JSON with an extra comma and therefore error out.

Also, one question, I was hoping you could answer. Does queryStream always return one row per stream.on "data" event?

Thanks!

[bchr02]

Never mind the question. I look it up (RTFM!) in the docs and confirmed that it's by row. 😀

[bchr02]

@dmcghan one thing that is not clear within the docs is whether the metadata event always fires prior to any data events firing. I am trying to avoid a race condition, so it's important to know this. Thank you.

[cjbj]

@bchr02 thanks for the update. Good to see you are still using node-oracledb. @dmcghan is heads down on some C19 work.

You can see where the metadata event is emitted in https://github.com/oracle/node-oracledb/blob/v5.2.0/lib/queryStream.js#L71

[bchr02]

Thanks @cjbj. Yea, I still like to get my hands dirty in code whenever I can get a chance.