Js data refactor builders

js/.gitignore

@@ -23,7 +23,8 @@ npm-debug.log*
 yarn-debug.log*
 yarn-error.log*
 
-.vscode
+.vscode/**
+!.vscode/launch.json
 
 # Runtime data
 pids

js/.vscode/launch.json

@@ -0,0 +1,155 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "type": "node",
+            "request": "launch",
+            "name": "Debug Gulp Build",
+            "program": "${workspaceFolder}/node_modules/gulp/bin/gulp.js",
+            "args": [
+                "build",
+                // Specify we want to debug the "src" target, which won't clean or build -- essentially a "dry-run" of the gulp build
+                "--target", "src"
+            ]
+        },
+        {
+            "type": "node",
+            "request": "launch",
+            "name": "Debug Unit Tests",
+            "cwd": "${workspaceRoot}",
+            "program": "${workspaceFolder}/node_modules/.bin/jest",
+            "skipFiles": [
+                "<node_internals>/**/*.js",
+                "${workspaceFolder}/node_modules/**/*.js"
+            ],
+            "env": {
+                "NODE_NO_WARNINGS": "1",
+                "READABLE_STREAM": "disable",
+                "TEST_DOM_STREAMS": "true",
+                "TEST_NODE_STREAMS": "true",
+                "TEST_FILE_NAMES": "primitive-empty primitive simple datetime decimal dictionary nested struct_example",
+                // Modify these environment variables to run tests on a specific compilation target + module format combo
+                "TEST_TS_SOURCE": "true",
+                // "TEST_TS_SOURCE": "false",
+                // "TEST_TARGET": "es5",
+                // "TEST_MODULE": "cjs"
+            },
+            "args": [
+                // "-i",
+                "test/unit/",
+                "test/integration/",
+
+                // Uncomment any of these to run individual test suites
+                // "test/unit/int-tests.ts",
+                // "test/unit/table-tests.ts",
+
+                // "test/unit/vector/vector-tests.ts",
+                // "test/unit/vector/bool-vector-tests.ts",
+                // "test/unit/vector/date-vector-tests.ts",
+                // "test/unit/vector/float16-vector-tests.ts",
+                // "test/unit/vector/numeric-vector-tests.ts",
+
+                // "test/unit/ipc/message-reader-tests.ts",
+                // "test/unit/ipc/reader/file-reader-tests.ts",
+                // "test/unit/ipc/reader/json-reader-tests.ts",
+                // "test/unit/ipc/reader/from-inference-tests.ts",
+                // "test/unit/ipc/reader/stream-reader-tests.ts",
+                // "test/unit/ipc/reader/streams-dom-tests.ts",
+                // "test/unit/ipc/reader/streams-node-tests.ts",
+                // "test/unit/ipc/writer/writer-tests.ts",
+                // "test/unit/ipc/writer/streams-dom-tests.ts",
+                // "test/unit/ipc/writer/streams-node-tests.ts",
+            ]
+        },
+        {
+            "type": "node",
+            "request": "launch",
+            "name": "Debug Integration Tests",
+            "cwd": "${workspaceRoot}",
+            "program": "${workspaceFolder}/node_modules/.bin/jest",
+            "skipFiles": [
+                "<node_internals>/**/*.js",
+                "${workspaceFolder}/node_modules/**/*.js"
+            ],
+            "env": {
+                "TEST_TS_SOURCE": "true",
+                "NODE_NO_WARNINGS": "1",
+                "READABLE_STREAM": "disable",
+                "TEST_DOM_STREAMS": "true",
+                "TEST_NODE_STREAMS": "true",
+                "TEST_FILE_NAMES": "primitive-empty primitive simple datetime decimal dictionary nested struct_example"
+            },
+            "args": [
+                // "-i",
+                "test/integration/*"
+            ]
+        },
+        {
+            "type": "node",
+            "request": "launch",
+            "name": "Debug bin/file-to-stream",
+            "env": { "ARROW_JS_DEBUG": "src", "TS_NODE_CACHE": "false" },
+            "runtimeArgs": ["-r", "ts-node/register"],
+            "skipFiles": [
+                "<node_internals>/**/*.js",
+                "${workspaceFolder}/node_modules/**/*.js"
+            ],
+            "args": [
+                "${workspaceFolder}/bin/file-to-stream.js",
+                "./test/data/cpp/file/struct_example.arrow",
+                "./struct_example-stream-out.arrow",
+            ]
+        },
+        {
+            "type": "node",
+            "request": "launch",
+            "name": "Debug bin/stream-to-file",
+            "env": { "ARROW_JS_DEBUG": "src", "TS_NODE_CACHE": "false" },
+            "runtimeArgs": ["-r", "ts-node/register"],
+            "skipFiles": [
+                "<node_internals>/**/*.js",
+                "${workspaceFolder}/node_modules/**/*.js"
+            ],
+            "args": [
+                "${workspaceFolder}/bin/stream-to-file.js",
+                "./test/data/cpp/stream/struct_example.arrow",
+                "./struct_example-file-out.arrow",
+            ]
+        },
+        {
+            "type": "node",
+            "request": "launch",
+            "name": "Debug bin/json-to-arrow",
+            "env": { "ARROW_JS_DEBUG": "src", "TS_NODE_CACHE": "false" },
+            "runtimeArgs": ["-r", "ts-node/register"],
+            "skipFiles": [
+                "<node_internals>/**/*.js",
+                "${workspaceFolder}/node_modules/**/*.js"
+            ],
+            "args": [
+                "${workspaceFolder}/bin/json-to-arrow.js",
+                "-j", "./test/data/json/struct_example.json",
+                "-a", "./struct_example-stream-out.arrow",
+                "-f", "stream"
+            ]
+        },
+        {
+            "type": "node",
+            "request": "launch",
+            "name": "Debug bin/print-buffer-alignment",
+            "env": { "ARROW_JS_DEBUG": "src", "TS_NODE_CACHE": "false" },
+            "runtimeArgs": ["-r", "ts-node/register"],
+            "skipFiles": [
+                "<node_internals>/**/*.js",
+                "${workspaceFolder}/node_modules/**/*.js"
+            ],
+            "args": [
+                "${workspaceFolder}/bin/print-buffer-alignment.js",
+                "./test/data/cpp/stream/struct_example.arrow"
+            ]
+        }
+    ]
+}

js/bin/file-to-stream.js

@@ -17,21 +17,24 @@
 // specific language governing permissions and limitations
 // under the License.
 
+// @ts-check
+
 const fs = require('fs');
 const path = require('path');
-
-const encoding = 'binary';
-const ext = process.env.ARROW_JS_DEBUG === 'src' ? '.ts' : '';
-const { util: { PipeIterator } } = require(`../index${ext}`);
-const { Table, serializeStream, fromReadableStream } = require(`../index${ext}`);
+const eos = require('util').promisify(require('stream').finished);
+const extension = process.env.ARROW_JS_DEBUG === 'src' ? '.ts' : '';
+const { RecordBatchReader, RecordBatchStreamWriter } = require(`../index${extension}`);
 
 (async () => {
-    // Todo (ptaylor): implement `serializeStreamAsync` that accepts an
-    // AsyncIterable<Buffer>, rather than aggregating into a Table first
-    const in_ = process.argv.length < 3
-        ? process.stdin : fs.createReadStream(path.resolve(process.argv[2]));
-    const out = process.argv.length < 4
-        ? process.stdout : fs.createWriteStream(path.resolve(process.argv[3]));
-    new PipeIterator(serializeStream(await Table.fromAsync(fromReadableStream(in_))), encoding).pipe(out);
+
+    const readable = process.argv.length < 3 ? process.stdin : fs.createReadStream(path.resolve(process.argv[2]));
+    const writable = process.argv.length < 4 ? process.stdout : fs.createWriteStream(path.resolve(process.argv[3]));
+
+    const fileToStream = readable
+        .pipe(RecordBatchReader.throughNode())
+        .pipe(RecordBatchStreamWriter.throughNode())
+        .pipe(writable);
+
+    await eos(fileToStream);
 
 })().catch((e) => { console.error(e); process.exit(1); });

js/bin/integration.js

@@ -17,60 +17,92 @@
 // specific language governing permissions and limitations
 // under the License.
 
+// @ts-check
+
 const fs = require('fs');
-const glob = require('glob');
 const path = require('path');
+const { promisify } = require('util');
+const glob = promisify(require('glob'));
+const { zip } = require('ix/iterable/zip');
 const child_process = require(`child_process`);
+const asyncDone = promisify(require('async-done'));
 const argv = require(`command-line-args`)(cliOpts(), { partial: true });
-const gulpPath = require.resolve(path.join(`..`, `node_modules/gulp/bin/gulp.js`));
-
-let jsonPaths = [...(argv.json || [])];
-let arrowPaths = [...(argv.arrow || [])];
 
-if (!argv.mode) {
-    return print_usage();
+const exists = async (p) => {
+    try {
+        return !!(await fs.promises.stat(p));
+    } catch (e) { return false; }
 }
 
-let mode = argv.mode.toUpperCase();
-if (mode === 'VALIDATE' && !jsonPaths.length) {
-    jsonPaths = glob.sync(path.resolve(__dirname, `../test/data/json/`, `*.json`));
-    if (!arrowPaths.length) {
-        [jsonPaths, arrowPaths] = jsonPaths.reduce(([jsonPaths, arrowPaths], jsonPath) => {
-            const { name } = path.parse(jsonPath);
-            for (const source of ['cpp', 'java']) {
-                for (const format of ['file', 'stream']) {
-                    const arrowPath = path.resolve(__dirname, `../test/data/${source}/${format}/${name}.arrow`);
-                    if (fs.existsSync(arrowPath)) {
-                        jsonPaths.push(jsonPath);
-                        arrowPaths.push(arrowPath);
-                    }
-                }
-            }
-            return [jsonPaths, arrowPaths];
-        }, [[], []]);
-        console.log(`jsonPaths: [\n\t${jsonPaths.join('\n\t')}\n]`);
-        console.log(`arrowPaths: [\n\t${arrowPaths.join('\n\t')}\n]`);
+(async () => {
+
+    if (!argv.mode) {
+        return print_usage();
+    }
+
+    let mode = argv.mode.toUpperCase();
+    let jsonPaths = [...(argv.json || [])];
+    let filePaths = [...(argv.file || [])];
+    let streamPaths = [...(argv.stream || [])];
+
+    if (mode === 'VALIDATE' && !jsonPaths.length) {
+        jsonPaths = await glob(path.resolve(__dirname, `../test/data/json/`, `*.json`));
+        if (!filePaths.length) {
+            [jsonPaths, filePaths] = await loadJSONAndArrowPaths(jsonPaths, 'cpp', 'file');
+            [jsonPaths, filePaths] = await loadJSONAndArrowPaths(jsonPaths, 'java', 'file');
+        }
+        if (!streamPaths.length) {
+            [jsonPaths, filePaths] = await loadJSONAndArrowPaths(jsonPaths, 'cpp', 'stream');
+            [jsonPaths, filePaths] = await loadJSONAndArrowPaths(jsonPaths, 'java', 'stream');
+        }
+        for (let [jsonPath, filePath, streamPath] of zip(jsonPaths, filePaths, streamPaths)) {
+            console.log(`jsonPath: ${jsonPath}`);
+            console.log(`filePath: ${filePath}`);
+            console.log(`streamPath: ${streamPath}`);
+        }
     }
-} else if (!jsonPaths.length) {
-    return print_usage();
-}
 
-switch (mode) {
-    case 'VALIDATE':
-        const args = [`test`, `-i`].concat(argv._unknown || []);
-        jsonPaths.forEach((p, i) => {
-            args.push('-j', p, '-a', arrowPaths[i]);
-        });
-        process.exitCode = child_process.spawnSync(
-            gulpPath, args,
-            {
+    if (!jsonPaths.length) {
+        return print_usage();
+    }
+
+    switch (mode) {
+        case 'VALIDATE':
+
+            const args = [`test`, `-i`].concat(argv._unknown || []);
+            const gulp = require.resolve(path.join(__dirname, `../node_modules/gulp/bin/gulp.js`));
+
+            for (let [jsonPath, filePath, streamPath] of zip(jsonPaths, filePaths, streamPaths)) {
+                args.push('-j', jsonPath, '-f', filePath, '-s', streamPath);
+            }
+
+            await asyncDone(() => child_process.spawn(gulp, args, {
                 cwd: path.resolve(__dirname, '..'),
                 stdio: ['ignore', 'inherit', 'inherit']
-            }
-        ).status || process.exitCode || 0;
-        break;
-    default:
-        print_usage();
+            }));
+
+            break;
+        default:
+            return print_usage();
+    }
+})()
+.then((x) => +x || 0, (e) => {
+    e && process.stderr.write(`${e}`);
+    return process.exitCode || 1;
+}).then((code) => process.exit(code));
+
+async function loadJSONAndArrowPaths(jsonPaths, source, format) {
+    const jPaths = [];
+    const aPaths = [];
+    for (const jsonPath of jsonPaths) {
+        const { name } = path.parse(jsonPath);
+        const arrowPath = path.resolve(__dirname, `../test/data/${source}/${format}/${name}.arrow`);
+        if (await exists(arrowPath)) {
+            jPaths.push(jsonPath);
+            aPaths.push(arrowPath);
+        }
+    }
+    return [jPaths, aPaths];
 }
 
 function cliOpts() {
@@ -80,6 +112,10 @@ function cliOpts() {
             name: 'mode',
             description: 'The integration test to run'
         },
+        { name: 'json',   alias: 'j', type: String, multiple: true, defaultValue: [], description: 'The JSON file[s] to read/write' },
+        { name: 'file',   alias: 'f', type: String, multiple: true, defaultValue: [], description: 'The Arrow file[s] to read/write' },
+        { name: 'stream', alias: 's', type: String, multiple: true, defaultValue: [], description: 'The Arrow stream[s] to read/write' },
+
         {
             type: String,
             name: 'arrow', alias: 'a',
@@ -118,5 +154,5 @@ function print_usage() {
             ]
         },
     ]));
-    process.exit(1);
+    return 1;
 }