Эффективное чтение, обработка и вставка данных с помощью PG-Promise и PG-Query-Stream

Я собираюсь сделать следующее.

1. Запросите большую таблицу с группой по запросу, чтобы выполнить суммирование значений.
2. Запустите эти записи через процедуру, чтобы добавить некоторые дополнительные данные.
3. Эффективно вставьте их в БД.

Я пытался сделать это, используя pg-query-stream, чтобы считывать данные в виде потока, а затем подсчитывать эти записи в пакетах, например. 1000 за раз, и как только мы достигнем предела пакетов, затем используйте pg-promise pgp.helpers.insert для вставки данных.

У меня проблема в том, что я не могу понять, как правильно приостановить поток, чтобы вставка завершилась, прежде чем продолжить. Особенно на on.end()

Код, который я пробовал, приведен ниже

const { performance } = require('perf_hooks')
const QueryStream = require('pg-query-stream')

const batchInsertData = (tenant, stream, records, insertColumnSet, options = {}) => {
  stream.pause()
  const t0 = performance.now()
  let query = tenant.db.$config.pgp.helpers.insert(records, insertColumnSet)

  if (options.onConflictExpression) {
    query += options.onConflictExpression
  }

  tenant.db.none(query)
    .then(() => {
      let t1 = performance.now()
      console.info('Inserted ' + records.length + ' records done in ' + ((t1 - t0) / 1000) + ' (seconds).')
      stream.resume()
    })
    .catch(error => {
      throw error
    })
}

module.exports = (tenant, sql, columnSet, recordMapper, options = {}) => {
  try {
    return new Promise((resolve, reject) => {
      const query = new QueryStream(sql)

      // Set options as required
      options.batchSize = parseInt(options.batchSize) || 1000
      options.onConflictExpression = options.onConflictExpression || null

      let records = []
      let batchNumber = 1
      let recordCount = 0

      let t0 = performance.now()
      tenant.db.stream(query, (stream) => {
        stream.on('data', (record) => {
          const mappedRecord = recordMapper(record)
          records.push(mappedRecord)
          recordCount++

          if (records.length === options.batchSize) {
            batchInsertData(tenant, stream, records, columnSet, options)
            records = []
            console.info(`Batch ${batchNumber} done`)
            batchNumber++
          }
        })
        stream.on('end', () => {
        // If any records are left that are not part of a batch insert here.
          if (records.length !== 0) {
            batchInsertData(tenant, stream, records, columnSet, options)
            records = []
            console.info(`Batch ${batchNumber} done`)
            batchNumber++
            console.info('Total Records: ' + recordCount)
            let t1 = performance.now()
            console.info('Duration:', ((t1 - t0) / 1000) + ' (seconds).')
          } else {
            console.info('Total Records: ' + recordCount)
            let t1 = performance.now()
            console.info('Duration:', ((t1 - t0) / 1000) + ' (seconds).')
          }
        })
        stream.on('error', (error) => {
          throw error
        })
      })
        .then(data => {
          resolve()
        })
        .catch(error => {
          console.info('ERROR:', error)
          reject(error)
        })
    })
  } catch (err) {
    throw err
  }
}

Я не уверен, что подход, который я пытаюсь, является лучшим. Я пробовал несколько разных вещей, основанных на документации, которую я могу найти по pg-promise и потокам, но не получил удовольствия.

Любая помощь/совет приветствуется.

Спасибо

Павел

Попытка 2

Ниже моя вторая попытка использовать getNextData и последовательность в соответствии со страницей импорта данных. Пытаюсь определить, как подключить к нему поток, чтобы извлекать только пакеты данных за раз перед вставкой.

const { performance } = require('perf_hooks')
const QueryStream = require('pg-query-stream')

module.exports = (tenant, sql, columnSet, recordMapper, options = {}) => {

  try {
    // Set options as required
    options.batchSize = parseInt(options.batchSize) || 1000
    options.onConflictExpression = options.onConflictExpression || null

    const query = new QueryStream(sql)

    function getNextData(transaction, index) {
      return new Promise(async (resolve, reject) => {
        if (index < options.batchSize) {
          let count = 1
          await transaction.stream(query, async (stream) => {
            let records = []
            await tenant.db.$config.pgp.spex.stream.read.call(transaction, stream, function (streamIndex, streamData) {  
              stream.resume()
              count++
              console.info(count, streamIndex, streamData)        

              records.push(streamData[0])

              if (records.length === options.batchSize) {
                stream.pause()
                resolve(records)
              }
            }, {readChunks: true})

          })  
        }
        resolve(null)
      })
    }

    return tenant.db.tx('massive-insert', (transaction) => {
      return transaction.sequence((index) => {          
        return getNextData(transaction, index)
          .then((records) => {
            if (records > 0) {
              let query = tenant.db.$config.pgp.helpers.insert(records, columnSet)

              if (options.onConflictExpression) {
                query += options.onConflictExpression
              }

              const i0 = performance.now()
              return transaction.none(query)
                .then(() => {
                  let i1 = performance.now()
                  console.info('Inserted ' + records.length + ' records done in ' + ((i1 - i0) / 1000) + ' (seconds).')
                })
            }
          })
      })
    })
  } catch (err) {
    throw err
  }
}