How do I measure progress while streaming a postgres database query via pg-query-stream?

Question

I have an ETL query which requires me to read a large number of rows and then apply some transformation on them and save them back to a separate table in Postgres. I am using pg-query-stream and I plan to run the test function below inside a bullmq job.

How do I measure progress (current rows processed / total number of rows) for the given stream below?

const { Pool } = require('pg');
const JSONStream = require('JSONStream');
const QueryStream = require('pg-query-stream');

function test() {
  const pool = new Pool({
    database: process.env.POSTGRES_DB,
    host: process.env.POSTGRES_HOST,
    port: +process.env.POSTGRES_PORT,
    password: process.env.POSTGRES_PASSWORD,
    ssl: process.env.POSTGRES_SSL === 'true',
    user: process.env.POSTGRES_USER,
  });
  const query = `
    SELECT 
        feed_item_id, 
        title, 
        summary, 
        content 
    FROM 
        feed_items 
    ORDER BY 
        pubdate DESC, 
        feed_item_id
    LIMIT 50
    `;

  pool.connect((err, client, done) => {
    if (err) throw err;
    const queryStream = new QueryStream(query, [], {
      batchSize: 200,
    });
    const stream = client.query(queryStream);
    console.log(stream);
    stream.pipe(JSONStream.stringify());
    stream.on('error', (error) => {
      console.error(error);
      done();
    });
    stream.on('end', () => {
      console.log('stream has ended');
      done();
    });
    stream.on('data', async (row) => {
      stream.pause();
      console.log('data received', row.feed_item_id);
      //   const progress = index / ???
      //   Simulate async task
      await timeout(10);
      stream.resume();
    });
  });
}

test();

Should be very easy via pg-iterator, because you can control every next-row read. — vitaly-t
– vitaly-t, Commented Nov 23, 2022 at 10:30
@vitaly-t i l definitely take a look into this one, thank you for the headsup — PirateApp
– PirateApp, Commented Nov 23, 2022 at 12:30

Pavel Bely · Accepted Answer · 2022-04-14 14:56:56Z

1

To do that you'll need to first get count of the rows in the table.
You can do it as described here.

And with that count you can pipe QueryStream into another stream and measure progress there:

const Stream = require('stream')

class ProgressStream extends Stream.Writable{
  constructor(total) {
    super();
    this.i = 0;
    this.total = total;
  }

  _write(chunk, enc, done) {
    this.i++;
    if (this.i % 100 === 1) {
      console.log(`Processed ${this.i} out of ${this.total} of rows`);
    }
    done();
  }
}

const progressStream = new ProgressStream(count);
progressStream.on('finish', () => {
  console.log(`All ${progressStream.i} rows processed!`)
});

stream.pipe(progressStream);

answered Apr 14, 2022 at 14:56

Pavel Bely

2,4751 gold badge20 silver badges28 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

How do I measure progress while streaming a postgres database query via pg-query-stream?

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related