export/import column map support

packages/api/src/shell/copyStream.js

@@ -1,18 +1,47 @@
 const EnsureStreamHeaderStream = require('../utility/EnsureStreamHeaderStream');
+const Stream = require('stream');
+const ColumnMapTransformStream = require('../utility/ColumnMapTransformStream');
+
+function copyStream(input, output, options) {
+  const { columns } = options || {};
+
+  const transforms = [];
+  if (columns) {
+    transforms.push(new ColumnMapTransformStream(columns));
+  }
+  if (output.requireFixedStructure) {
+    transforms.push(new EnsureStreamHeaderStream());
+  }
+
+  // return new Promise((resolve, reject) => {
+  //   Stream.pipeline(input, ...transforms, output, err => {
+  //     if (err) {
+  //       reject(err);
+  //     } else {
+  //       resolve();
+  //     }
+  //   });
+  // });
 
-function copyStream(input, output) {
   return new Promise((resolve, reject) => {
     const finisher = output['finisher'] || output;
     finisher.on('finish', resolve);
     finisher.on('error', reject);
 
-    if (output.requireFixedStructure) {
-      const ensureHeader = new EnsureStreamHeaderStream();
-      input.pipe(ensureHeader);
-      ensureHeader.pipe(output);
-    } else {
-      input.pipe(output);
+    let lastStream = input;
+    for (const tran of transforms) {
+      lastStream.pipe(tran);
+      lastStream = tran;
     }
+    lastStream.pipe(output);
+
+    //  if (output.requireFixedStructure) {
+    //   const ensureHeader = new EnsureStreamHeaderStream();
+    //   input.pipe(ensureHeader);
+    //   ensureHeader.pipe(output);
+    // } else {
+    //   input.pipe(output);
+    // }
   });
 }
 

packages/api/src/utility/ColumnMapTransformStream.js

@@ -0,0 +1,21 @@
+const stream = require('stream');
+const { transformRowUsingColumnMap } = require('dbgate-tools');
+
+class ColumnMapTransformStream extends stream.Transform {
+  constructor(columns) {
+    super({ objectMode: true });
+    this.columns = columns;
+  }
+  _transform(chunk, encoding, done) {
+    if (chunk.__isStreamHeader) {
+      // skip stream header
+      done();
+      return;
+    }
+
+    this.push(transformRowUsingColumnMap(chunk, this.columns));
+    done();
+  }
+}
+
+module.exports = ColumnMapTransformStream;