diff --git a/AGENTS.md b/AGENTS.md index fafd4d020a..c699d7a5ee 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,6 +32,7 @@ - Never abbreviate variables, always type out the full name in camelCase (variables, functions, fields), PascalCase (types), CAPITAL_CASE (constant) - Add TSDoc to all new classes, functions, methods, fields. - Prefer verbNoun structure for function and method names. +- Prefer `readonly` fields over getters for fixed or constructor-derived values. - We end JavaScript statements with semicolons. Do not remove semicolons. ## Notes diff --git a/docs/modules/compression/README.md b/docs/modules/compression/README.md index 7b308f3a83..a66cadd7c5 100644 --- a/docs/modules/compression/README.md +++ b/docs/modules/compression/README.md @@ -7,6 +7,32 @@ The `@loaders.gl/compression` module provides a selection of lossless, compression/decompression "transforms" with a unified interface that work both in browsers and in Node.js +For async code that only needs decompression, the lightweight +[`@loaders.gl/compression/native-decompression`](/docs/modules/compression/api-reference/native-decompression) +entrypoint probes the runtime's +`DecompressionStream` implementation for gzip, deflate, raw deflate, Brotli, and Zstandard. The +entrypoint has no codec imports, so supported runtimes do not pull fallback codec code into the +initial bundle. It returns `null` when the runtime or exact format is unavailable, allowing callers +to load a fallback only when needed. +From-v5.0 + +```typescript +import {decompressWithNativeDecompressionStream} from '@loaders.gl/compression/native-decompression'; + +async function decompressGzip(input: ArrayBuffer): Promise { + const output = await decompressWithNativeDecompressionStream(input, 'gzip'); + if (output) { + return output; + } + const {GZipCompression} = await import('@loaders.gl/compression/gzip-compression'); + return new GZipCompression().decompress(input); +} +``` + +Parquet and SPZ parsing use this lightweight path automatically before lazily loading their +codec-backed fallbacks. Existing compression classes keep their deterministic codec behavior for +compression and synchronous decompression. + ## API | Compression Class | Format | Characteristics | Library Size | Notes | diff --git a/docs/modules/compression/api-reference/brotli-compression.md b/docs/modules/compression/api-reference/brotli-compression.md index 34e41ea924..2db22b0234 100644 --- a/docs/modules/compression/api-reference/brotli-compression.md +++ b/docs/modules/compression/api-reference/brotli-compression.md @@ -13,3 +13,5 @@ Implements the [`Compression](./compression) API. ## Methods ### `constructor(options?: object)` + +`options` is optional when using the built-in Brotli decoder. diff --git a/docs/modules/compression/api-reference/native-decompression.md b/docs/modules/compression/api-reference/native-decompression.md new file mode 100644 index 0000000000..493472080b --- /dev/null +++ b/docs/modules/compression/api-reference/native-decompression.md @@ -0,0 +1,38 @@ +# Native Decompression + +The lightweight `@loaders.gl/compression/native-decompression` entrypoint exposes async +decompression through the runtime's `DecompressionStream` API without importing fallback codecs. +It supports `gzip`, `deflate`, `deflate-raw`, `brotli`, and forward-compatible `zstd` +constructor probing. +From-v5.0 + +The helpers return `null` only when `DecompressionStream` or the requested format is unavailable. +After a native stream is created, decompression errors are propagated to the caller. + +```typescript +import { + decompressWithNativeDecompressionStream +} from '@loaders.gl/compression/native-decompression'; + +async function decompressGzip(compressedData: ArrayBuffer): Promise { + const output = await decompressWithNativeDecompressionStream(compressedData, 'gzip'); + if (output) { + return output; + } + const {GZipCompression} = await import('@loaders.gl/compression/gzip-compression'); + return new GZipCompression().decompress(compressedData); +} +``` + +## Functions + +### `decompressWithNativeDecompressionStream(input, format)` + +Decompresses one `ArrayBuffer` and returns an exact `ArrayBuffer`, or `null` when the runtime +does not support the requested format. + +### `decompressBatchesWithNativeDecompressionStream(inputBatches, format)` + +Creates an incremental native decompression stream for iterable or async iterable `ArrayBuffer` +batches. It returns an async iterable of exact `ArrayBuffer` chunks, or `null` when the runtime +does not support the requested format. diff --git a/docs/modules/compression/api-reference/zstd-compression.md b/docs/modules/compression/api-reference/zstd-compression.md index 60a97efb5c..a4910604fc 100644 --- a/docs/modules/compression/api-reference/zstd-compression.md +++ b/docs/modules/compression/api-reference/zstd-compression.md @@ -6,6 +6,11 @@ Compresses / decompresses Zstandard encoded data. +Inject `zstd-codec` through `options.modules` for compression and decompression through this +codec-backed class. Async-only callers can probe future native Zstandard support through the +lightweight `@loaders.gl/compression/native-decompression` entrypoint without importing +`zstd-codec`. + ## Interface Implements the [`Compression](./compression) API. @@ -13,3 +18,6 @@ Implements the [`Compression](./compression) API. ## Methods ### `constructor(options?: object)` + +`options` is optional at construction time. Supply `{modules: {'zstd-codec': ZstdCodec}}` before +calling compression or decompression methods. diff --git a/docs/modules/parquet/api-reference/parquet-loader.md b/docs/modules/parquet/api-reference/parquet-loader.md index 5c00766614..46e486ca89 100644 --- a/docs/modules/parquet/api-reference/parquet-loader.md +++ b/docs/modules/parquet/api-reference/parquet-loader.md @@ -119,8 +119,12 @@ field-level GeoArrow metadata. ## Compressions -Some compressions are big and need to be imported explicitly by the application -and passed to the `ParquetLoader` +Async Parquet parsing first probes the runtime's native `DecompressionStream` for gzip, Brotli, +and Zstandard pages. Those probes come from a lightweight entrypoint with no codec imports, and +codec-backed implementations are loaded only when native support is unavailable. Native Zstandard +support is not yet widely available, so inject `zstd-codec` for broad compatibility; when +provided, it takes precedence over the native path. LZ4 still requires `lz4js`. +From-v5.0 ```typescript import {ParquetLoader} from '@loaders.gl/parquet'; diff --git a/docs/modules/splats/api-reference/spz-loader.md b/docs/modules/splats/api-reference/spz-loader.md index 1cecfc49ab..965e617e63 100644 --- a/docs/modules/splats/api-reference/spz-loader.md +++ b/docs/modules/splats/api-reference/spz-loader.md @@ -17,9 +17,15 @@ ## Usage -SPZ version 4 uses ZSTD-compressed attribute streams. Inject `zstd-codec` through loader options so applications that only use `SPLATLoader` or `KSPLATLoader` do not pay the ZSTD dependency cost. +SPZ version 4 uses ZSTD-compressed attribute streams. Async SPZ parsing first probes the lightweight +native decompression entrypoint, which has no codec imports, before lazily loading the +codec-backed fallback. Native Zstandard support is not yet widely available, so inject +`zstd-codec` through loader options for broad runtime compatibility; when provided, it takes +precedence over the native path. +From-v5.0 ```typescript +// Install zstd-codec for broad runtime compatibility. // npm install @loaders.gl/core @loaders.gl/splats zstd-codec import {load} from '@loaders.gl/core'; @@ -72,4 +78,4 @@ Schema metadata includes `loaders_gl.semantic_type = gaussian-splats` and `loade | Option | Type | Default | Description | | -------------- | --------------- | --------------- | ---------------------------------------- | | `splats.shape` | `'arrow-table'` | `'arrow-table'` | Selects Mesh Arrow table output. V1 only supports `arrow-table`. | -| `modules` | `object` | `{}` | Must include `{'zstd-codec': ZstdCodec}` to decode SPZ version 4 streams. | +| `modules` | `object` | `{}` | Include `{'zstd-codec': ZstdCodec}` for broad SPZ version 4 runtime compatibility. | diff --git a/docs/whats-new.mdx b/docs/whats-new.mdx index beb0c98325..909b6c5e7d 100644 --- a/docs/whats-new.mdx +++ b/docs/whats-new.mdx @@ -73,6 +73,10 @@ Release Date: 2026 - `preload(loader)` now returns and caches parser-bearing implementations for unbundled loaders. - `preloadSync(loader)` NEW returns a cached parser-bearing implementation, or `null` if the loader has not been preloaded. +**@loaders.gl/compression** + +- The new lightweight `@loaders.gl/compression/native-decompression` entrypoint probes runtime support for gzip, deflate, raw deflate, Brotli, and future Zstandard decompression without importing codec fallbacks. Parquet and SPZ use it before lazily loading codec-backed implementations; until native Zstandard support becomes widely available, most runtimes still need `zstd-codec` for Zstandard-backed parsing. + **@loaders.gl/deck-layers** - `SplatLayer` NEW - renders GraphDECO-style Gaussian splat Arrow tables from `PLYLoader` or `@loaders.gl/splats`. diff --git a/modules/compression/package.json b/modules/compression/package.json index 2d8af71633..e3bf43c0d7 100644 --- a/modules/compression/package.json +++ b/modules/compression/package.json @@ -30,6 +30,38 @@ "import": "./dist/index.js", "require": "./dist/index.cjs" }, + "./native-decompression": { + "types": "./dist/native-decompression.d.ts", + "import": "./dist/native-decompression.js" + }, + "./no-compression": { + "types": "./dist/no-compression.d.ts", + "import": "./dist/no-compression.js" + }, + "./gzip-compression": { + "types": "./dist/gzip-compression.d.ts", + "import": "./dist/gzip-compression.js" + }, + "./deflate-compression": { + "types": "./dist/deflate-compression.d.ts", + "import": "./dist/deflate-compression.js" + }, + "./brotli-compression": { + "types": "./dist/brotli-compression.d.ts", + "import": "./dist/brotli-compression.js" + }, + "./snappy-compression": { + "types": "./dist/snappy-compression.d.ts", + "import": "./dist/snappy-compression.js" + }, + "./lz4-compression": { + "types": "./dist/lz4-compression.d.ts", + "import": "./dist/lz4-compression.js" + }, + "./zstd-compression": { + "types": "./dist/zstd-compression.d.ts", + "import": "./dist/zstd-compression.js" + }, "./compression-worker.js": { "import": "./dist/compression-worker.js" }, @@ -63,13 +95,8 @@ "pako": "1.0.11", "snappyjs": "^0.6.1" }, - "optionalDependencies": { - "@types/brotli": "^1.3.0", - "brotli": "^1.3.2", - "lz4js": "^0.2.0", - "zstd-codec": "^0.1" - }, "devDependencies": { + "@types/brotli": "^1.3.0", "brotli": "^1.3.2", "lz4js": "^0.2.0", "zstd-codec": "^0.1" diff --git a/modules/compression/src/brotli-compression.ts b/modules/compression/src/brotli-compression.ts new file mode 100644 index 0000000000..ae15bfd654 --- /dev/null +++ b/modules/compression/src/brotli-compression.ts @@ -0,0 +1,6 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export {BrotliCompression} from './lib/brotli-compression'; +export type {BrotliCompressionOptions} from './lib/brotli-compression'; diff --git a/modules/compression/src/deflate-compression.ts b/modules/compression/src/deflate-compression.ts new file mode 100644 index 0000000000..d2a12fafbf --- /dev/null +++ b/modules/compression/src/deflate-compression.ts @@ -0,0 +1,6 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export {DeflateCompression} from './lib/deflate-compression'; +export type {DeflateCompressionOptions} from './lib/deflate-compression'; diff --git a/modules/compression/src/gzip-compression.ts b/modules/compression/src/gzip-compression.ts new file mode 100644 index 0000000000..2884be18e1 --- /dev/null +++ b/modules/compression/src/gzip-compression.ts @@ -0,0 +1,6 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export {GZipCompression} from './lib/gzip-compression'; +export type {GZipCompressionOptions} from './lib/gzip-compression'; diff --git a/modules/compression/src/lib/brotli-compression.ts b/modules/compression/src/lib/brotli-compression.ts index 3e22af80d5..9db0c368b5 100644 --- a/modules/compression/src/lib/brotli-compression.ts +++ b/modules/compression/src/lib/brotli-compression.ts @@ -48,7 +48,7 @@ export class BrotliCompression extends Compression { readonly isSupported = true; readonly options: BrotliCompressionOptions; - constructor(options: BrotliCompressionOptions) { + constructor(options: BrotliCompressionOptions = {}) { super(options); this.options = options; registerJSModules(options?.modules); diff --git a/modules/compression/src/lib/decompression-stream.ts b/modules/compression/src/lib/decompression-stream.ts new file mode 100644 index 0000000000..b877496752 --- /dev/null +++ b/modules/compression/src/lib/decompression-stream.ts @@ -0,0 +1,170 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +/** + * Compression formats that runtimes may expose through DecompressionStream. + * + * Brotli and zstd are included even when the installed TypeScript DOM definitions + * have not caught up with runtime support. + */ +export type NativeDecompressionFormat = 'brotli' | 'deflate' | 'deflate-raw' | 'gzip' | 'zstd'; + +type NativeDecompressionStreamConstructor = new ( + format: NativeDecompressionFormat +) => DecompressionStream; + +/** + * Decompresses one ArrayBuffer with a runtime-provided DecompressionStream. + * + * @param input Compressed input data. + * @param format Compression format to decode. + * @returns Decompressed data, or null when the runtime does not support the format. + */ +export async function decompressWithNativeDecompressionStream( + input: ArrayBuffer, + format: NativeDecompressionFormat +): Promise { + const outputBatches = decompressBatchesWithNativeDecompressionStream([input], format); + return outputBatches ? await concatenateNativeDecompressionBatches(outputBatches) : null; +} + +/** + * Decompresses batches with a runtime-provided DecompressionStream. + * + * @param inputBatches Compressed input data. + * @param format Compression format to decode. + * @returns Decompressed batches, or null when the runtime does not support the format. + */ +export function decompressBatchesWithNativeDecompressionStream( + inputBatches: AsyncIterable | Iterable, + format: NativeDecompressionFormat +): AsyncIterable | null { + const decompressionStream = createNativeDecompressionStream(format); + return decompressionStream + ? transformBatchesWithNativeDecompressionStream(inputBatches, decompressionStream) + : null; +} + +/** + * Creates a runtime-provided DecompressionStream when the requested format is supported. + * + * @param format Compression format to decode. + * @returns A native decompression stream, or null when it is unavailable. + */ +function createNativeDecompressionStream( + format: NativeDecompressionFormat +): DecompressionStream | null { + if (typeof globalThis.DecompressionStream === 'undefined') { + return null; + } + // Node's zlib-backed implementation dereferences the global Buffer internally. + if (globalThis.process?.versions?.node && typeof globalThis.Buffer === 'undefined') { + return null; + } + + try { + const DecompressionStreamConstructor = + globalThis.DecompressionStream as unknown as NativeDecompressionStreamConstructor; + return new DecompressionStreamConstructor(format); + } catch (error) { + if (error instanceof TypeError || (error as Error)?.name === 'TypeError') { + return null; + } + throw error; + } +} + +/** + * Pipes compressed batches into a native stream while yielding decompressed output incrementally. + * + * @param inputBatches Compressed input data. + * @param decompressionStream Native stream that performs decompression. + * @yields Exact ArrayBuffer views of decompressed output chunks. + */ +async function* transformBatchesWithNativeDecompressionStream( + inputBatches: AsyncIterable | Iterable, + decompressionStream: DecompressionStream +): AsyncIterable { + const writer = decompressionStream.writable.getWriter(); + const reader = decompressionStream.readable.getReader(); + const writePromise = writeBatchesToNativeDecompressionStream(inputBatches, writer); + writePromise.catch(() => {}); + let outputCompleted = false; + + try { + while (true) { + const {done, value} = await reader.read(); + if (done) { + outputCompleted = true; + break; + } + yield copyExactArrayBuffer(value); + } + await writePromise; + } finally { + if (!outputCompleted) { + await reader.cancel().catch(() => {}); + await writer.abort().catch(() => {}); + await writePromise.catch(() => {}); + } + reader.releaseLock(); + } +} + +/** + * Concatenates decompressed batches without importing codec-adjacent utilities. + * + * @param outputBatches Decompressed output batches. + * @returns One exact ArrayBuffer containing every output byte. + */ +async function concatenateNativeDecompressionBatches( + outputBatches: AsyncIterable +): Promise { + const batches: Uint8Array[] = []; + let byteLength = 0; + for await (const outputBatch of outputBatches) { + const bytes = new Uint8Array(outputBatch); + batches.push(bytes); + byteLength += bytes.byteLength; + } + + const output = new Uint8Array(byteLength); + let byteOffset = 0; + for (const batch of batches) { + output.set(batch, byteOffset); + byteOffset += batch.byteLength; + } + return output.buffer; +} + +/** + * Copies one stream chunk into an exact ArrayBuffer. + * + * @param value Decompressed stream chunk. + * @returns Exact ArrayBuffer containing only the chunk bytes. + */ +function copyExactArrayBuffer(value: Uint8Array): ArrayBuffer { + return new Uint8Array(value).buffer; +} + +/** + * Writes compressed batches to a native decompression stream and closes its input. + * + * @param inputBatches Compressed input data. + * @param writer Native stream writer receiving compressed batches. + */ +async function writeBatchesToNativeDecompressionStream( + inputBatches: AsyncIterable | Iterable, + writer: WritableStreamDefaultWriter +): Promise { + try { + for await (const inputBatch of inputBatches) { + await writer.write(new Uint8Array(inputBatch)); + } + await writer.close(); + } catch (error) { + await writer.abort(error).catch(() => {}); + throw error; + } +} diff --git a/modules/compression/src/lib/zstd-compression.ts b/modules/compression/src/lib/zstd-compression.ts index daa876937a..399e3cd1d3 100644 --- a/modules/compression/src/lib/zstd-compression.ts +++ b/modules/compression/src/lib/zstd-compression.ts @@ -34,7 +34,7 @@ export class ZstdCompression extends Compression { * zstd-codec is an injectable dependency due to big size * @param options */ - constructor(options: CompressionOptions) { + constructor(options: CompressionOptions = {}) { super(options); this.options = options; registerJSModules(options?.modules); diff --git a/modules/compression/src/lz4-compression.ts b/modules/compression/src/lz4-compression.ts new file mode 100644 index 0000000000..a3c44c4858 --- /dev/null +++ b/modules/compression/src/lz4-compression.ts @@ -0,0 +1,5 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export {LZ4Compression} from './lib/lz4-compression'; diff --git a/modules/compression/src/native-decompression.ts b/modules/compression/src/native-decompression.ts new file mode 100644 index 0000000000..c45f4dcd29 --- /dev/null +++ b/modules/compression/src/native-decompression.ts @@ -0,0 +1,9 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export { + decompressBatchesWithNativeDecompressionStream, + decompressWithNativeDecompressionStream +} from './lib/decompression-stream'; +export type {NativeDecompressionFormat} from './lib/decompression-stream'; diff --git a/modules/compression/src/no-compression.ts b/modules/compression/src/no-compression.ts new file mode 100644 index 0000000000..37db42930f --- /dev/null +++ b/modules/compression/src/no-compression.ts @@ -0,0 +1,5 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export {NoCompression} from './lib/no-compression'; diff --git a/modules/compression/src/snappy-compression.ts b/modules/compression/src/snappy-compression.ts new file mode 100644 index 0000000000..4fa93c5b56 --- /dev/null +++ b/modules/compression/src/snappy-compression.ts @@ -0,0 +1,5 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export {SnappyCompression} from './lib/snappy-compression'; diff --git a/modules/compression/src/zstd-compression.ts b/modules/compression/src/zstd-compression.ts new file mode 100644 index 0000000000..08cf653910 --- /dev/null +++ b/modules/compression/src/zstd-compression.ts @@ -0,0 +1,5 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +export {ZstdCompression} from './lib/zstd-compression'; diff --git a/modules/compression/test/compression.spec.ts b/modules/compression/test/compression.spec.ts index 88e97bb266..eca604bfd5 100644 --- a/modules/compression/test/compression.spec.ts +++ b/modules/compression/test/compression.spec.ts @@ -15,9 +15,20 @@ import { // LZOCompression, CompressionWorker } from '@loaders.gl/compression'; +import { + decompressBatchesWithNativeDecompressionStream, + decompressWithNativeDecompressionStream +} from '@loaders.gl/compression/native-decompression'; import {processOnWorker, isBrowser, WorkerFarm} from '@loaders.gl/worker-utils'; import {concatenateArrayBuffers, concatenateArrayBuffersAsync} from '@loaders.gl/loader-utils'; import {getData, compareArrayBuffers} from './utils/test-utils'; +import { + installRecordingDecompressionStream, + NATIVE_DECOMPRESSION_FIXTURES, + NATIVE_DECOMPRESSION_TEST_DATA, + supportsNativeDecompressionStream, + type NativeDecompressionTestFormat +} from './utils/native-decompression-test-utils'; // Import big dependencies @@ -162,6 +173,141 @@ test('compression#batched', async t => { t.end(); }); +test('native decompression#real DecompressionStream formats', async t => { + for (const format of Object.keys( + NATIVE_DECOMPRESSION_FIXTURES + ) as NativeDecompressionTestFormat[]) { + if (!(await supportsNativeDecompressionStream(format))) { + t.comment(`${format} DecompressionStream is not available in this runtime`); + continue; + } + + const nativeFormats: NativeDecompressionTestFormat[] = []; + const restoreDecompressionStream = installRecordingDecompressionStream(nativeFormats); + + try { + const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES[format]).buffer; + + const decompressedData = await decompressWithNativeDecompressionStream( + compressedData, + format + ); + t.ok( + decompressedData && compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), + `native atomic ${format} decompression works` + ); + + const splitIndex = Math.max(1, Math.floor(compressedData.byteLength / 2)); + const compressedBatches = [ + compressedData.slice(0, splitIndex), + compressedData.slice(splitIndex, compressedData.byteLength) + ]; + const decompressedBatches = decompressBatchesWithNativeDecompressionStream( + compressedBatches, + format + ); + t.ok(decompressedBatches, `native batched ${format} stream is created`); + const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); + t.ok( + compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedBatchData), + `native batched ${format} decompression works` + ); + t.deepEqual( + nativeFormats, + [format, format], + `${format} uses the native stream for atomic and batched decompression` + ); + } finally { + restoreDecompressionStream(); + } + } + + t.end(); +}); + +test('native decompression#mocked zstd atomic and batched', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream({ + formats, + supportedFormats: ['zstd'] + }); + + try { + const inputBatches = [new Uint8Array([1, 2, 3]).buffer, new Uint8Array([4, 5, 6]).buffer]; + const inputData = concatenateArrayBuffers(...inputBatches); + + const decompressedData = await decompressWithNativeDecompressionStream(inputData, 'zstd'); + t.ok( + decompressedData && compareArrayBuffers(inputData, decompressedData), + 'native atomic zstd needs no codec' + ); + + const decompressedBatches = decompressBatchesWithNativeDecompressionStream( + inputBatches, + 'zstd' + ); + t.ok(decompressedBatches, 'native batched zstd stream is created'); + const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); + t.ok( + compareArrayBuffers(inputData, decompressedBatchData), + 'native batched zstd needs no codec' + ); + t.deepEqual(formats, ['zstd', 'zstd'], 'zstd maps to the native zstd format'); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + +test('native decompression#unsupported formats return null', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream({ + formats, + supportedFormats: [] + }); + + try { + const inputData = new Uint8Array([1, 2, 3]).buffer; + t.equal( + await decompressWithNativeDecompressionStream(inputData, 'zstd'), + null, + 'atomic unsupported format returns null' + ); + t.equal( + decompressBatchesWithNativeDecompressionStream([inputData], 'zstd'), + null, + 'batched unsupported format returns null' + ); + t.deepEqual(formats, ['zstd', 'zstd'], 'both paths probe the requested format'); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + +test('native decompression#stream failures propagate', async t => { + const restoreDecompressionStream = installMockDecompressionStream({ + formats: [], + supportedFormats: ['zstd'], + failWith: new Error('mock native decompression failed') + }); + + try { + const inputData = new Uint8Array([1, 2, 3]).buffer; + await t.rejects( + decompressWithNativeDecompressionStream(inputData, 'zstd'), + /mock native decompression failed/, + 'native stream errors propagate' + ); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + // WORKER TESTS test('gzip#worker', async t => { const {binaryData} = getData(); @@ -260,3 +406,74 @@ test.skip('zstd#worker', async t => { t.ok(compareArrayBuffers(decompressdData, binaryData), 'compress/decompress level 6'); t.end(); }); + +type MockDecompressionStreamOptions = { + formats: string[]; + supportedFormats: string[]; + failWith?: Error; +}; + +/** + * Installs a deterministic DecompressionStream double and returns a restorer. + * + * @param options Mock formats and failure behavior. + * @returns Callback that restores the original global constructor. + */ +function installMockDecompressionStream(options: MockDecompressionStreamOptions): () => void { + const originalDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'DecompressionStream'); + + class MockDecompressionStream { + readonly readable: ReadableStream; + readonly writable: WritableStream; + + /** Creates a mock native decompression stream for supported formats. */ + constructor(format: string) { + options.formats.push(format); + if (!options.supportedFormats.includes(format)) { + throw new TypeError('mock compression format is unsupported'); + } + + const transformStream = new TransformStream({ + transform(chunk, controller) { + if (options.failWith) { + throw options.failWith; + } + controller.enqueue(copyBufferSource(chunk)); + } + }); + this.readable = transformStream.readable; + this.writable = transformStream.writable; + } + } + + Object.defineProperty(globalThis, 'DecompressionStream', { + configurable: true, + writable: true, + value: MockDecompressionStream + }); + + return () => { + if (originalDescriptor) { + Object.defineProperty(globalThis, 'DecompressionStream', originalDescriptor); + } else { + delete (globalThis as any).DecompressionStream; + } + }; +} + +/** + * Copies a native stream input chunk into a Uint8Array. + * + * @param bufferSource Native stream input chunk. + * @returns Copied bytes for the mock stream output. + */ +function copyBufferSource(bufferSource: BufferSource): Uint8Array { + if (bufferSource instanceof ArrayBuffer) { + return new Uint8Array(bufferSource).slice(); + } + return new Uint8Array( + bufferSource.buffer, + bufferSource.byteOffset, + bufferSource.byteLength + ).slice(); +} diff --git a/modules/compression/test/decompression-stream.node.spec.ts b/modules/compression/test/decompression-stream.node.spec.ts new file mode 100644 index 0000000000..076980ae30 --- /dev/null +++ b/modules/compression/test/decompression-stream.node.spec.ts @@ -0,0 +1,90 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import test from 'tape-promise/tape'; +import { + decompressBatchesWithNativeDecompressionStream, + decompressWithNativeDecompressionStream +} from '@loaders.gl/compression/native-decompression'; +import {concatenateArrayBuffersAsync} from '@loaders.gl/loader-utils'; +import {compareArrayBuffers} from './utils/test-utils'; +import { + installRecordingDecompressionStream, + NATIVE_DECOMPRESSION_FIXTURES, + NATIVE_DECOMPRESSION_TEST_DATA, + supportsNativeDecompressionStream, + type NativeDecompressionTestFormat +} from './utils/native-decompression-test-utils'; + +type MutableGlobalThis = typeof globalThis & { + Buffer?: typeof Buffer; +}; + +test('native decompression#real DecompressionStream formats in Node.js', async t => { + for (const format of Object.keys( + NATIVE_DECOMPRESSION_FIXTURES + ) as NativeDecompressionTestFormat[]) { + if (!(await supportsNativeDecompressionStream(format))) { + t.comment(`${format} DecompressionStream is not available in this runtime`); + continue; + } + + const nativeFormats: NativeDecompressionTestFormat[] = []; + const restoreDecompressionStream = installRecordingDecompressionStream(nativeFormats); + + try { + const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES[format]).buffer; + + const decompressedData = await decompressWithNativeDecompressionStream( + compressedData, + format + ); + t.ok( + decompressedData && compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), + `native atomic ${format} decompression works in Node.js` + ); + + const splitIndex = Math.max(1, Math.floor(compressedData.byteLength / 2)); + const compressedBatches = [ + compressedData.slice(0, splitIndex), + compressedData.slice(splitIndex, compressedData.byteLength) + ]; + const decompressedBatches = decompressBatchesWithNativeDecompressionStream( + compressedBatches, + format + ); + t.ok(decompressedBatches, `native batched ${format} stream is created in Node.js`); + const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); + t.ok( + compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedBatchData), + `native batched ${format} decompression works in Node.js` + ); + t.deepEqual( + nativeFormats, + [format, format], + `${format} uses the native stream for atomic and batched Node.js decompression` + ); + } finally { + restoreDecompressionStream(); + } + } + + t.end(); +}); + +test('native decompression#returns null without global Buffer in Node.js', async t => { + const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES.gzip).buffer; + const mutableGlobalThis = globalThis as MutableGlobalThis; + const originalBuffer = mutableGlobalThis.Buffer; + mutableGlobalThis.Buffer = undefined; + + try { + const decompressedData = await decompressWithNativeDecompressionStream(compressedData, 'gzip'); + t.equal(decompressedData, null, 'native helper lets callers choose a fallback without Buffer'); + } finally { + mutableGlobalThis.Buffer = originalBuffer; + } + + t.end(); +}); diff --git a/modules/compression/test/utils/native-decompression-test-utils.ts b/modules/compression/test/utils/native-decompression-test-utils.ts new file mode 100644 index 0000000000..60d449543c --- /dev/null +++ b/modules/compression/test/utils/native-decompression-test-utils.ts @@ -0,0 +1,82 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +/** Native formats supported by the DecompressionStream adapter. */ +export type NativeDecompressionTestFormat = 'brotli' | 'deflate' | 'deflate-raw' | 'gzip' | 'zstd'; + +/** Uncompressed bytes shared by real runtime DecompressionStream format tests. */ +export const NATIVE_DECOMPRESSION_TEST_DATA = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]) + .buffer as ArrayBuffer; + +/** Fixed compressed fixtures for every format supported by the native adapter. */ +export const NATIVE_DECOMPRESSION_FIXTURES: Record = { + gzip: [ + 31, 139, 8, 0, 0, 0, 0, 0, 0, 19, 99, 100, 98, 102, 97, 101, 99, 231, 224, 4, 0, 158, 171, 239, + 64, 9, 0, 0, 0 + ], + deflate: [120, 156, 99, 100, 98, 102, 97, 101, 99, 231, 224, 4, 0, 0, 174, 0, 46], + 'deflate-raw': [99, 100, 98, 102, 97, 101, 99, 231, 224, 4, 0], + brotli: [11, 4, 128, 1, 2, 3, 4, 5, 6, 7, 8, 9, 3], + zstd: [40, 181, 47, 253, 32, 9, 73, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9] +}; + +/** + * Returns whether the runtime accepts one native DecompressionStream format. + * + * @param format Native format to probe. + * @returns Whether the runtime can construct a stream for the format. + */ +export async function supportsNativeDecompressionStream( + format: NativeDecompressionTestFormat +): Promise { + if (typeof globalThis.DecompressionStream === 'undefined') { + return false; + } + + try { + const DecompressionStreamConstructor = globalThis.DecompressionStream as unknown as new ( + format: NativeDecompressionTestFormat + ) => DecompressionStream; + const decompressionStream = new DecompressionStreamConstructor(format); + await decompressionStream.writable.abort(); + return true; + } catch { + return false; + } +} + +/** + * Installs a constructor wrapper that records formats while delegating to the real runtime API. + * + * @param formats Mutable list that receives every requested native format. + * @returns Callback that restores the original global constructor. + */ +export function installRecordingDecompressionStream( + formats: NativeDecompressionTestFormat[] +): () => void { + const originalDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'DecompressionStream'); + const DecompressionStreamConstructor = globalThis.DecompressionStream as unknown as new ( + format: NativeDecompressionTestFormat + ) => DecompressionStream; + const RecordingDecompressionStream = function ( + format: NativeDecompressionTestFormat + ): DecompressionStream { + formats.push(format); + return new DecompressionStreamConstructor(format); + } as unknown as typeof DecompressionStream; + + Object.defineProperty(globalThis, 'DecompressionStream', { + configurable: true, + writable: true, + value: RecordingDecompressionStream + }); + + return () => { + if (originalDescriptor) { + Object.defineProperty(globalThis, 'DecompressionStream', originalDescriptor); + } else { + delete (globalThis as any).DecompressionStream; + } + }; +} diff --git a/modules/parquet/package.json b/modules/parquet/package.json index 8c8bdef12b..046b6c09d3 100644 --- a/modules/parquet/package.json +++ b/modules/parquet/package.json @@ -92,7 +92,6 @@ "@loaders.gl/wkt": "5.0.0-alpha.1", "@probe.gl/log": "^4.1.1", "async-mutex": "^0.2.2", - "brotli": "^1.3.2", "isomorphic-ws": "^5.0.0", "lz4js": "^0.2.0", "node-int64": "^0.4.0", @@ -101,15 +100,16 @@ "snappyjs": "^0.6.0", "thrift": "^0.24.0", "util": "^0.12.5", - "varint": "^6.0.0", - "zstd-codec": "^0.1" + "varint": "^6.0.0" }, "devDependencies": { "@types/node-int64": "^0.4.29", "@types/thrift": "^0.10.8", "@types/varint": "^5.0.0", + "brotli": "^1.3.2", "hyparquet": "1.27.1", - "hyparquet-compressors": "1.1.1" + "hyparquet-compressors": "1.1.1", + "zstd-codec": "^0.1" }, "peerDependencies": { "@loaders.gl/core": "~5.0.0-alpha.0", diff --git a/modules/parquet/src/parquetjs/compression.ts b/modules/parquet/src/parquetjs/compression.ts index b6d4481505..a8bda4b5a8 100644 --- a/modules/parquet/src/parquetjs/compression.ts +++ b/modules/parquet/src/parquetjs/compression.ts @@ -5,78 +5,63 @@ // Forked from https://github.com/kbajalc/parquets under MIT license // Forked from https://github.com/ironSource/parquetjs under MIT license +import type {Compression} from '@loaders.gl/compression'; import { - Compression, - NoCompression, - GZipCompression, - SnappyCompression, - BrotliCompression, - // LZOCompression, - LZ4Compression, - ZstdCompression -} from '@loaders.gl/compression'; -import {registerJSModules} from '@loaders.gl/loader-utils'; + decompressWithNativeDecompressionStream, + type NativeDecompressionFormat +} from '@loaders.gl/compression/native-decompression'; +import {getJSModuleOrNull, registerJSModules} from '@loaders.gl/loader-utils'; import {ParquetCompression} from './schema/declare'; import {toArrayBuffer, toUint8Array} from './utils/binary-utils'; -// TODO switch to worker compression to avoid bundling... - -// import brotli from 'brotli'; - brotli has problems with decompress in browsers -// import brotliDecompress from 'brotli/decompress'; -import lz4js from 'lz4js'; -// import lzo from 'lzo'; -// import {ZstdCodec} from 'zstd-codec'; - -// Inject large dependencies through Compression constructor options -const modules = { - // brotli has problems with decompress in browsers - // brotli: { - // decompress: brotliDecompress, - // compress: () => { - // throw new Error('brotli compress'); - // } - // }, - lz4js - // lzo - // 'zstd-codec': ZstdCodec -}; - /** * See https://github.com/apache/parquet-format/blob/master/Compression.md */ -// @ts-expect-error -export const PARQUET_COMPRESSION_METHODS: Record = { - UNCOMPRESSED: new NoCompression(), - GZIP: new GZipCompression(), - SNAPPY: new SnappyCompression(), - BROTLI: new BrotliCompression({modules}), - // TODO: Understand difference between LZ4 and LZ4_RAW - LZ4: new LZ4Compression({modules}), - LZ4_RAW: new LZ4Compression({modules}), - // - // LZO: new LZOCompression({modules}), - ZSTD: new ZstdCompression({modules}) +export const PARQUET_COMPRESSION_METHODS: Partial> = { + UNCOMPRESSED: true, + GZIP: true, + SNAPPY: true, + BROTLI: true, + // TODO: Understand difference between LZ4 and LZ4_RAW. + LZ4: true, + LZ4_RAW: true, + ZSTD: true }; +/** Native formats available to asynchronous Parquet page decompression. */ +const PARQUET_NATIVE_DECOMPRESSION_FORMATS: Partial< + Record +> = { + GZIP: 'gzip', + BROTLI: 'brotli', + ZSTD: 'zstd' +}; + +/** Lazily constructed codec-backed compression implementations. */ +const compressionPromises: Partial>> = {}; + /** - * Register compressions that have big external libraries - * @param options.modules External library dependencies + * Registers optional codec modules without eagerly loading codec-backed implementations. + * + * @param options.modules External library dependencies. */ export async function preloadCompressions(options?: {modules?: {[key: string]: any}}) { registerJSModules(options?.modules); - const compressions = Object.values(PARQUET_COMPRESSION_METHODS); - return await Promise.all(compressions.map(compression => compression.preload(options?.modules))); } /** * Deflate a value using compression method `method` */ export async function deflate(method: ParquetCompression, value: Uint8Array): Promise { - const compression = PARQUET_COMPRESSION_METHODS[method]; - if (!compression) { + if (!(method in PARQUET_COMPRESSION_METHODS)) { throw new Error(`parquet: invalid compression method: ${method}`); } + if (method === 'UNCOMPRESSED') { + return value; + } + + const compression = await getParquetCompression(method); const inputArrayBuffer = toArrayBuffer(value); const compressedArrayBuffer = await compression.compress(inputArrayBuffer); return toUint8Array(compressedArrayBuffer); @@ -90,11 +75,88 @@ export async function decompress( value: Uint8Array, size: number ): Promise { - const compression = PARQUET_COMPRESSION_METHODS[method]; - if (!compression) { + if (!(method in PARQUET_COMPRESSION_METHODS)) { throw new Error(`parquet: invalid compression method: ${method}`); } const inputArrayBuffer = toArrayBuffer(value); + if (method === 'UNCOMPRESSED') { + return toUint8Array(inputArrayBuffer); + } + + const nativeFormat = PARQUET_NATIVE_DECOMPRESSION_FORMATS[method]; + if (nativeFormat && shouldUseNativeDecompressionStream(method)) { + const nativeOutput = await decompressWithNativeDecompressionStream( + inputArrayBuffer, + nativeFormat + ); + if (nativeOutput) { + return toUint8Array(nativeOutput); + } + } + + const compression = await getParquetCompression(method); const compressedArrayBuffer = await compression.decompress(inputArrayBuffer, size); return toUint8Array(compressedArrayBuffer); } + +/** + * Returns whether a native stream can take precedence for one Parquet compression method. + * + * @param method Parquet compression method. + * @returns Whether no explicitly registered codec should take precedence. + */ +function shouldUseNativeDecompressionStream(method: ParquetCompression): boolean { + if (method === 'BROTLI') { + return !getJSModuleOrNull('brotli'); + } + if (method === 'ZSTD') { + return !getJSModuleOrNull('zstd-codec'); + } + return true; +} + +/** + * Loads one codec-backed implementation only after native decompression is unavailable. + * + * @param method Parquet compression method. + * @returns Codec-backed compression implementation. + */ +async function getParquetCompression(method: ParquetCompression): Promise { + compressionPromises[method] ||= createParquetCompression(method); + return await compressionPromises[method]; +} + +/** + * Creates one lazily loaded codec-backed Parquet compression implementation. + * + * @param method Parquet compression method. + * @returns Codec-backed compression implementation. + */ +async function createParquetCompression(method: ParquetCompression): Promise { + switch (method) { + case 'GZIP': { + const {GZipCompression} = await import('@loaders.gl/compression/gzip-compression'); + return new GZipCompression(); + } + case 'SNAPPY': { + const {SnappyCompression} = await import('@loaders.gl/compression/snappy-compression'); + return new SnappyCompression(); + } + case 'BROTLI': { + const {BrotliCompression} = await import('@loaders.gl/compression/brotli-compression'); + return new BrotliCompression(); + } + case 'LZ4': + case 'LZ4_RAW': { + const {LZ4Compression} = await import('@loaders.gl/compression/lz4-compression'); + const lz4js = getJSModuleOrNull('lz4js') || (await import('lz4js')).default; + return new LZ4Compression({modules: {lz4js}}); + } + case 'ZSTD': { + const {ZstdCompression} = await import('@loaders.gl/compression/zstd-compression'); + return new ZstdCompression(); + } + default: + throw new Error(`parquet: invalid compression method: ${method}`); + } +} diff --git a/modules/parquet/test/init.ts b/modules/parquet/test/init.ts index 88c748565e..60871ac916 100644 --- a/modules/parquet/test/init.ts +++ b/modules/parquet/test/init.ts @@ -24,7 +24,6 @@ const modules = { 'zstd-codec': ZstdCodec }; -// Start loading compression modules in the background to minimize -// time spent during test case execution +// Register compression modules used by tests. // eslint-disable-next-line @typescript-eslint/no-misused-promises preloadCompressions({modules}); diff --git a/modules/parquet/test/parquet-compression.spec.ts b/modules/parquet/test/parquet-compression.spec.ts new file mode 100644 index 0000000000..cae0000d39 --- /dev/null +++ b/modules/parquet/test/parquet-compression.spec.ts @@ -0,0 +1,193 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import test from 'tape-promise/tape'; +import {ZstdCodec} from 'zstd-codec'; +import {ZstdCompression} from '@loaders.gl/compression/zstd-compression'; +import {decompress} from '../src/parquetjs/compression'; + +test('Parquet compression#native streams avoid codec fallbacks', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream(formats, [ + 'gzip', + 'brotli', + 'zstd' + ]); + const restoreBrotli = removeRegisteredModule('brotli'); + const restoreZstd = removeRegisteredModule('zstd-codec'); + const input = new Uint8Array([1, 2, 3, 4]); + + try { + for (const [method, format] of [ + ['GZIP', 'gzip'], + ['BROTLI', 'brotli'], + ['ZSTD', 'zstd'] + ] as const) { + const output = await decompress(method, input, input.byteLength); + t.deepEqual([...output], [...input], `${method} uses native decompression`); + t.equal(formats.at(-1), format, `${method} maps to ${format}`); + } + } finally { + restoreZstd(); + restoreBrotli(); + restoreDecompressionStream(); + } + + t.end(); +}); + +test('Parquet compression#provided modules bypass native streams', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream(formats, ['brotli', 'zstd']); + const input = new Uint8Array([1, 2, 3, 4]); + const restoreBrotli = replaceRegisteredModule('brotli', { + decompress: (bytes: Uint8Array) => bytes, + compress: () => { + throw new Error('compression is not used by this test'); + } + }); + const restoreZstd = replaceRegisteredModule('zstd-codec', ZstdCodec); + + try { + const brotliOutput = await decompress('BROTLI', input, input.byteLength); + t.deepEqual([...brotliOutput], [...input], 'provided Brotli module decompresses data'); + + const compression = new ZstdCompression({modules: {'zstd-codec': ZstdCodec}}); + await compression.preload(); + const compressedZstd = new Uint8Array(compression.compressSync(input.buffer)); + const zstdOutput = await decompress('ZSTD', compressedZstd, input.byteLength); + t.deepEqual([...zstdOutput], [...input], 'provided zstd-codec decompresses data'); + t.deepEqual(formats, [], 'provided modules bypass native stream probing'); + } finally { + restoreZstd(); + restoreBrotli(); + restoreDecompressionStream(); + } + + t.end(); +}); + +test('Parquet compression#unsupported native gzip lazily falls back to pako', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream(formats, []); + const compressedGzip = new Uint8Array([ + 31, 139, 8, 0, 0, 0, 0, 0, 0, 19, 99, 100, 98, 102, 97, 101, 99, 231, 224, 4, 0, 158, 171, + 239, 64, 9, 0, 0, 0 + ]); + + try { + const output = await decompress('GZIP', compressedGzip, 9); + t.deepEqual([...output], [1, 2, 3, 4, 5, 6, 7, 8, 9], 'pako fallback decompresses'); + t.deepEqual(formats, ['gzip'], 'native gzip is probed before lazy fallback'); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + +/** + * Installs a pass-through DecompressionStream double and returns a restorer. + * + * @param formats Mutable list receiving requested formats. + * @param supportedFormats Formats accepted by the mock constructor. + * @returns Callback that restores the original global constructor. + */ +function installMockDecompressionStream( + formats: string[], + supportedFormats: string[] +): () => void { + const originalDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'DecompressionStream'); + + class MockDecompressionStream { + readonly readable: ReadableStream; + readonly writable: WritableStream; + + /** Creates a pass-through stream for one supported format. */ + constructor(format: string) { + formats.push(format); + if (!supportedFormats.includes(format)) { + throw new TypeError('mock compression format is unsupported'); + } + const transformStream = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(copyBufferSource(chunk)); + } + }); + this.readable = transformStream.readable; + this.writable = transformStream.writable; + } + } + + Object.defineProperty(globalThis, 'DecompressionStream', { + configurable: true, + writable: true, + value: MockDecompressionStream + }); + + return () => { + if (originalDescriptor) { + Object.defineProperty(globalThis, 'DecompressionStream', originalDescriptor); + } else { + delete (globalThis as any).DecompressionStream; + } + }; +} + +/** + * Removes one registered injectable module and returns a restorer. + * + * @param moduleName Registered module name. + * @returns Callback that restores the original registration. + */ +function removeRegisteredModule(moduleName: string): () => void { + return replaceRegisteredModule(moduleName, undefined); +} + +/** + * Replaces one registered injectable module and returns a restorer. + * + * @param moduleName Registered module name. + * @param module Module value, or undefined to remove it. + * @returns Callback that restores the original registration. + */ +function replaceRegisteredModule(moduleName: string, module: unknown): () => void { + const globalWithLoaders = globalThis as any; + globalWithLoaders.loaders ||= {}; + const loaders = globalWithLoaders.loaders; + loaders.modules ||= {}; + const registeredModules = loaders.modules; + const hadModule = Object.prototype.hasOwnProperty.call(registeredModules, moduleName); + const originalModule = registeredModules[moduleName]; + if (module === undefined) { + delete registeredModules[moduleName]; + } else { + registeredModules[moduleName] = module; + } + + return () => { + if (hadModule) { + registeredModules[moduleName] = originalModule; + } else { + delete registeredModules[moduleName]; + } + }; +} + +/** + * Copies a native stream input chunk into a Uint8Array. + * + * @param bufferSource Native stream input chunk. + * @returns Copied bytes for the mock stream output. + */ +function copyBufferSource(bufferSource: BufferSource): Uint8Array { + if (bufferSource instanceof ArrayBuffer) { + return new Uint8Array(bufferSource).slice(); + } + return new Uint8Array( + bufferSource.buffer, + bufferSource.byteOffset, + bufferSource.byteLength + ).slice(); +} diff --git a/modules/parquet/test/parquet.bench.ts b/modules/parquet/test/parquet.bench.ts index 3363c5f698..c744f6122e 100644 --- a/modules/parquet/test/parquet.bench.ts +++ b/modules/parquet/test/parquet.bench.ts @@ -2,36 +2,113 @@ // SPDX-License-Identifier: MIT // Copyright (c) vis.gl contributors -import {ParquetJSLoader, ParquetLoader, GeoParquetLoader} from '@loaders.gl/parquet'; -import {fetchFile, load} from '@loaders.gl/core'; +import {GeoParquetLoader, ParquetJSLoader, ParquetLoader} from '@loaders.gl/parquet'; +import {fetchFile, load, parse, preload} from '@loaders.gl/core'; +import type {LoaderWithParser} from '@loaders.gl/loader-utils'; +import type {ObjectRowTable} from '@loaders.gl/schema'; +import {parquetReadObjects} from 'hyparquet'; +import {compressors} from 'hyparquet-compressors'; -// const PARQUET_URL = '@loaders.gl/parquet/test/data/apache/good/alltypes_plain.parquet'; const PARQUET_URL = '@loaders.gl/parquet/test/data/fruits.parquet'; +const LZ4_PARQUET_URL = + '@loaders.gl/parquet/test/data/apache/good/lz4_raw_compressed_larger.parquet'; +const HADOOP_LZ4_PARQUET_URL = + '@loaders.gl/parquet/test/data/apache/good/hadoop_lz4_compressed_larger.parquet'; +const DELTA_BYTE_ARRAY_PARQUET_URL = + '@loaders.gl/parquet/test/data/apache/good/delta_byte_array.parquet'; const GEO_PARQUET_URL = '@loaders.gl/parquet/test/data/geoparquet/airports.parquet'; -const IMPLEMENTATIONS = ['js', 'wasm'] as const; +const BENCHMARK_OPTIONS = {minIterations: 5, unit: 'rows'}; +const BENCHMARK_WARMUP_ITERATIONS = 2; + +type ParquetBenchmarkScenario = { + /** Human-readable fixture and projection label. */ + name: string; + /** Complete Parquet object held outside the timed benchmark callback. */ + arrayBuffer: ArrayBuffer; + /** Optional top-level Parquet columns decoded by every included implementation. */ + columns?: string[]; + /** Implementations included when a backend cannot correctly execute the scenario. */ + implementationIds?: ParquetBenchmarkImplementationId[]; +}; + +type ParquetBenchmarkImplementationId = 'typescript' | 'wasm' | 'hyparquet'; + +type ParquetBenchmarkImplementation = { + /** Stable implementation identifier used for scenario selection. */ + id: ParquetBenchmarkImplementationId; + /** Human-readable implementation and version label. */ + name: string; + /** Hot decode operation returning the validated output row count. */ + decode: (scenario: ParquetBenchmarkScenario) => Promise; +}; export async function parquetBench(suite) { - suite = suite.group('ParquetLoader'); - - let response = await fetchFile(PARQUET_URL); - const arrayBuffer = await response.arrayBuffer(); - - response = await fetchFile(GEO_PARQUET_URL); - const geoArrayBuffer = await response.arrayBuffer(); - - for (const implementation of IMPLEMENTATIONS) { - const loader = implementation === 'js' ? ParquetJSLoader : ParquetLoader; - suite.addAsync( - `load(${implementation === 'js' ? 'ParquetJSLoader' : 'ParquetLoader'}) - Parquet load`, - {multiplier: 40000, unit: 'rows'}, - async () => { - await load(arrayBuffer, loader, { - core: {worker: false} - }); - } - ); + const [ + parquetResponse, + lz4ParquetResponse, + hadoopLz4ParquetResponse, + deltaByteArrayParquetResponse, + geoParquetResponse + ] = await Promise.all([ + fetchFile(PARQUET_URL), + fetchFile(LZ4_PARQUET_URL), + fetchFile(HADOOP_LZ4_PARQUET_URL), + fetchFile(DELTA_BYTE_ARRAY_PARQUET_URL), + fetchFile(GEO_PARQUET_URL) + ]); + const [arrayBuffer, lz4ArrayBuffer, hadoopLz4ArrayBuffer, deltaByteArrayBuffer, geoArrayBuffer] = + await Promise.all([ + parquetResponse.arrayBuffer(), + lz4ParquetResponse.arrayBuffer(), + hadoopLz4ParquetResponse.arrayBuffer(), + deltaByteArrayParquetResponse.arrayBuffer(), + geoParquetResponse.arrayBuffer() + ]); + const [typescriptLoader, wasmLoader] = await Promise.all([ + preload(ParquetJSLoader, {core: {worker: false}}), + preload(ParquetLoader, {core: {worker: false}, parquet: {backend: 'wasm'}}) + ]); + const implementations = createParquetBenchmarkImplementations(typescriptLoader, wasmLoader); + const scenarios: ParquetBenchmarkScenario[] = [ + {name: 'LZ4_RAW full table', arrayBuffer: lz4ArrayBuffer}, + {name: 'Hadoop LZ4 full table', arrayBuffer: hadoopLz4ArrayBuffer}, + {name: 'DELTA_BYTE_ARRAY full table', arrayBuffer: deltaByteArrayBuffer}, + { + name: 'DELTA_BYTE_ARRAY projected columns', + arrayBuffer: deltaByteArrayBuffer, + columns: ['c_customer_id', 'c_email_address'], + // parquet-wasm 0.7.2 currently returns mismatched IPC schema/vector counts for this projection. + implementationIds: ['typescript', 'hyparquet'] + } + ]; + + for (const scenario of scenarios) { + const scenarioImplementations = scenario.implementationIds + ? implementations.filter(implementation => + scenario.implementationIds?.includes(implementation.id) + ) + : implementations; + const rowCount = await validateParquetBenchmarkScenario(scenario, scenarioImplementations); + suite = suite.groupSorted(`Parquet object-row decode - ${scenario.name}`); + + for (const implementation of scenarioImplementations) { + suite.addAsync( + `${implementation.name} - ${scenario.name}`, + {...BENCHMARK_OPTIONS, multiplier: rowCount}, + async () => { + const decodedRowCount = await implementation.decode(scenario); + if (decodedRowCount !== rowCount) { + throw new Error( + `${implementation.name} decoded ${decodedRowCount} rows; expected ${rowCount}` + ); + } + } + ); + } } + suite = suite.group('ParquetLoader Arrow'); + suite.addAsync( "load(ParquetLoader, shape: 'arrow-table') - Parquet load", {multiplier: 40000, unit: 'rows'}, @@ -83,3 +160,79 @@ export async function parquetBench(suite) { // }); // }); } + +/** Creates equivalent object-row decode cases for the maintained Parquet implementations. */ +function createParquetBenchmarkImplementations( + typescriptLoader: LoaderWithParser, + wasmLoader: LoaderWithParser +): ParquetBenchmarkImplementation[] { + return [ + { + id: 'typescript', + name: 'loaders.gl TypeScript', + decode: scenario => decodeWithLoadersGl(scenario, typescriptLoader, 'typescript') + }, + { + id: 'wasm', + name: 'loaders.gl parquet-wasm', + decode: scenario => decodeWithLoadersGl(scenario, wasmLoader, 'wasm') + }, + { + id: 'hyparquet', + name: 'hyparquet 1.27.1', + decode: decodeWithHyparquet + } + ]; +} + +/** Decodes one scenario through a preloaded loaders.gl implementation. */ +async function decodeWithLoadersGl( + scenario: ParquetBenchmarkScenario, + loader: LoaderWithParser, + backend: 'typescript' | 'wasm' +): Promise { + const table = (await parse(scenario.arrayBuffer, loader, { + core: {worker: false}, + parquet: {backend, columns: scenario.columns} + })) as ObjectRowTable; + return table.data.length; +} + +/** Decodes one scenario through the latest pinned hyparquet implementation. */ +async function decodeWithHyparquet(scenario: ParquetBenchmarkScenario): Promise { + const rows = await parquetReadObjects({ + file: scenario.arrayBuffer, + columns: scenario.columns, + compressors + }); + return rows.length; +} + +/** Warms every implementation and verifies that benchmark throughput uses a common row count. */ +async function validateParquetBenchmarkScenario( + scenario: ParquetBenchmarkScenario, + implementations: ParquetBenchmarkImplementation[] +): Promise { + const rowCounts: number[] = []; + for (const implementation of implementations) { + let rowCount = 0; + for (let iteration = 0; iteration < BENCHMARK_WARMUP_ITERATIONS; iteration++) { + rowCount = await implementation.decode(scenario); + } + rowCounts.push(rowCount); + } + + const expectedRowCount = rowCounts[0]; + for ( + let implementationIndex = 1; + implementationIndex < implementations.length; + implementationIndex++ + ) { + if (rowCounts[implementationIndex] !== expectedRowCount) { + throw new Error( + `${implementations[implementationIndex].name} decoded ${rowCounts[implementationIndex]} rows from ${scenario.name}; expected ${expectedRowCount}` + ); + } + } + return expectedRowCount; +} diff --git a/modules/splats/src/lib/parse-spz.ts b/modules/splats/src/lib/parse-spz.ts index 6109d0190b..9009b94c98 100644 --- a/modules/splats/src/lib/parse-spz.ts +++ b/modules/splats/src/lib/parse-spz.ts @@ -2,7 +2,8 @@ // SPDX-License-Identifier: MIT // Copyright (c) vis.gl contributors -import {ZstdCompression} from '@loaders.gl/compression'; +import {decompressWithNativeDecompressionStream} from '@loaders.gl/compression/native-decompression'; +import {getJSModuleOrNull, registerJSModules} from '@loaders.gl/loader-utils'; import type {MeshArrowTable} from '@loaders.gl/schema'; import type {GaussianSplats, SplatsLoaderOptions} from '../types'; import {makeGaussianSplatsArrowTable} from './splats-arrow-table'; @@ -51,10 +52,24 @@ export async function parseSPZToGaussianSplats( const compressedStreams = streamInfos.map(streamInfo => data.slice(streamInfo.compressedOffset, streamInfo.compressedOffset + streamInfo.compressedSize) ); - const compression = new ZstdCompression({modules: options?.modules}); + registerJSModules(options?.modules); + const useNativeDecompressionStream = !getJSModuleOrNull('zstd-codec'); + let compressionPromise: + | Promise<{ + decompress(input: ArrayBuffer, size?: number): Promise; + }> + | undefined; const streams = await Promise.all( compressedStreams.map((compressedStream, streamIndex) => - compression.decompress(compressedStream, streamInfos[streamIndex].uncompressedSize) + decompressSPZStream( + compressedStream, + streamInfos[streamIndex].uncompressedSize, + useNativeDecompressionStream, + () => { + compressionPromise ||= createZstdCompression(options?.modules); + return compressionPromise; + } + ) ) ); @@ -62,6 +77,45 @@ export async function parseSPZToGaussianSplats( return decodeSPZStreams(header, streams, data); } +/** + * Decompresses one SPZ stream natively when available, then lazily loads the codec fallback. + * + * @param compressedStream Compressed Zstandard stream. + * @param uncompressedSize Expected uncompressed byte length. + * @param useNativeDecompressionStream Whether no provided codec should take precedence. + * @param getCompression Lazily resolves the codec-backed Zstandard implementation. + * @returns Decompressed stream bytes. + */ +async function decompressSPZStream( + compressedStream: ArrayBuffer, + uncompressedSize: number, + useNativeDecompressionStream: boolean, + getCompression: () => Promise<{ + decompress(input: ArrayBuffer, size?: number): Promise; + }> +): Promise { + if (useNativeDecompressionStream) { + const nativeOutput = await decompressWithNativeDecompressionStream(compressedStream, 'zstd'); + if (nativeOutput) { + return nativeOutput; + } + } + + const compression = await getCompression(); + return await compression.decompress(compressedStream, uncompressedSize); +} + +/** + * Dynamically loads the codec-backed Zstandard class for unsupported native runtimes. + * + * @param modules Optional registered codec modules. + * @returns Codec-backed Zstandard implementation. + */ +async function createZstdCompression(modules?: {[key: string]: any}) { + const {ZstdCompression} = await import('@loaders.gl/compression/zstd-compression'); + return new ZstdCompression({modules}); +} + /** Parses and validates the SPZ v4 plaintext header. */ function parseSPZHeader(data: ArrayBuffer): SPZHeader { if (data.byteLength < SPZ_HEADER_BYTE_LENGTH) { diff --git a/modules/splats/test/spz-loader.spec.ts b/modules/splats/test/spz-loader.spec.ts index 5991ddab76..10c175c5ef 100644 --- a/modules/splats/test/spz-loader.spec.ts +++ b/modules/splats/test/spz-loader.spec.ts @@ -13,30 +13,60 @@ const modules = {'zstd-codec': ZstdCodec}; test('SPZLoader parses Niantic Spatial v4 Gaussian splats', async t => { const data = await makeSPZFixture(); - const table = await parse(data, SPZLoader, {modules}); - - t.equal(table.shape, 'arrow-table', 'returns MeshArrowTable'); - t.equal(table.topology, 'point-list', 'returns point-list topology'); - t.equal(table.data.numRows, 2, 'parses row count'); - t.equal( - table.data.schema.metadata.get('loaders_gl.gaussian_splats.source_format'), - 'spz', - 'adds source format metadata' - ); - t.deepEqual(table.data.getChild('POSITION')?.get(0)?.toArray(), [1, 2, -3], 'parses position'); - t.ok(Math.abs(Number(table.data.getChild('scale_1')?.get(0)) - 1) < 1e-6, 'decodes scale'); - t.ok( - Math.abs(Number(table.data.getChild('opacity')?.get(1)) - 64 / 255) < 1e-6, - 'decodes linear opacity' - ); - t.ok( - Math.abs(Number(table.data.getChild('f_dc_0')?.get(0)) - (140 / 255 - 0.5) / 0.15) < 1e-6, - 'decodes SPZ DC coefficient' - ); - t.ok(Math.abs(Number(table.data.getChild('rot_0')?.get(0)) - 1) < 1e-6, 'decodes rotation'); + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream(formats); + + try { + const table = await parse(data, SPZLoader, {modules}); + + t.equal(table.shape, 'arrow-table', 'returns MeshArrowTable'); + t.equal(table.topology, 'point-list', 'returns point-list topology'); + t.equal(table.data.numRows, 2, 'parses row count'); + t.equal( + table.data.schema.metadata.get('loaders_gl.gaussian_splats.source_format'), + 'spz', + 'adds source format metadata' + ); + t.deepEqual(table.data.getChild('POSITION')?.get(0)?.toArray(), [1, 2, -3], 'parses position'); + t.ok(Math.abs(Number(table.data.getChild('scale_1')?.get(0)) - 1) < 1e-6, 'decodes scale'); + t.ok( + Math.abs(Number(table.data.getChild('opacity')?.get(1)) - 64 / 255) < 1e-6, + 'decodes linear opacity' + ); + t.ok( + Math.abs(Number(table.data.getChild('f_dc_0')?.get(0)) - (140 / 255 - 0.5) / 0.15) < 1e-6, + 'decodes SPZ DC coefficient' + ); + t.ok(Math.abs(Number(table.data.getChild('rot_0')?.get(0)) - 1) < 1e-6, 'decodes rotation'); + + const directTable = await SPZLoaderWithParser.parse(data, {modules}); + t.equal(directTable.data.numRows, 2, 'parser subpath supports async parse'); + t.deepEqual(formats, [], 'provided zstd-codec bypasses native stream probing'); + } finally { + restoreDecompressionStream(); + } + t.end(); +}); + +test('SPZLoader uses native zstd without a codec module', async t => { + const data = await makeSPZFixture(false); + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream(formats); + const restoreZstd = removeRegisteredModule('zstd-codec'); + + try { + const table = await SPZLoaderWithParser.parse(data); + t.equal(table.data.numRows, 2, 'native zstd path parses SPZ data'); + t.deepEqual( + formats, + ['zstd', 'zstd', 'zstd', 'zstd', 'zstd'], + 'all SPZ streams use native zstd' + ); + } finally { + restoreZstd(); + restoreDecompressionStream(); + } - const directTable = await SPZLoaderWithParser.parse(data, {modules}); - t.equal(directTable.data.numRows, 2, 'parser subpath supports async parse'); t.end(); }); @@ -57,8 +87,13 @@ test('SPZLoader validates header', async t => { t.end(); }); -/** Builds a deterministic two-row SPZ v4 fixture with compressed streams. */ -async function makeSPZFixture(): Promise { +/** + * Builds a deterministic two-row SPZ v4 fixture. + * + * @param compressStreams Whether to encode fixture streams with zstd-codec. + * @returns SPZ fixture data. + */ +async function makeSPZFixture(compressStreams = true): Promise { const streams = [ makePositionStream(), new Uint8Array([128, 64]), @@ -66,11 +101,14 @@ async function makeSPZFixture(): Promise { makeScaleStream(), makeRotationStream() ]; - const compression = new ZstdCompression({modules}); - await compression.preload(modules); - const compressedStreams = streams.map( - stream => new Uint8Array(compression.compressSync(stream.buffer)) - ); + let compressedStreams = streams; + if (compressStreams) { + const compression = new ZstdCompression({modules}); + await compression.preload(modules); + compressedStreams = streams.map( + stream => new Uint8Array(compression.compressSync(stream.buffer)) + ); + } const headerByteLength = 32; const tocByteLength = compressedStreams.length * 16; const byteLength = @@ -102,6 +140,92 @@ async function makeSPZFixture(): Promise { return data; } +/** + * Installs a pass-through native zstd stream and returns a restorer. + * + * @param formats Mutable list receiving requested formats. + * @returns Callback that restores the original global constructor. + */ +function installMockDecompressionStream(formats: string[]): () => void { + const originalDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'DecompressionStream'); + + class MockDecompressionStream { + readonly readable: ReadableStream; + readonly writable: WritableStream; + + /** Creates a pass-through stream for the native zstd format. */ + constructor(format: string) { + formats.push(format); + if (format !== 'zstd') { + throw new TypeError('mock compression format is unsupported'); + } + const transformStream = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(copyBufferSource(chunk)); + } + }); + this.readable = transformStream.readable; + this.writable = transformStream.writable; + } + } + + Object.defineProperty(globalThis, 'DecompressionStream', { + configurable: true, + writable: true, + value: MockDecompressionStream + }); + + return () => { + if (originalDescriptor) { + Object.defineProperty(globalThis, 'DecompressionStream', originalDescriptor); + } else { + delete (globalThis as any).DecompressionStream; + } + }; +} + +/** + * Removes one registered injectable module and returns a restorer. + * + * @param moduleName Registered module name. + * @returns Callback that restores the original registration. + */ +function removeRegisteredModule(moduleName: string): () => void { + const globalWithLoaders = globalThis as any; + globalWithLoaders.loaders ||= {}; + const loaders = globalWithLoaders.loaders; + loaders.modules ||= {}; + const registeredModules = loaders.modules; + const hadModule = Object.prototype.hasOwnProperty.call(registeredModules, moduleName); + const originalModule = registeredModules[moduleName]; + delete registeredModules[moduleName]; + + return () => { + if (hadModule) { + registeredModules[moduleName] = originalModule; + } else { + delete registeredModules[moduleName]; + } + }; +} + +/** + * Copies a native stream input chunk into a Uint8Array. + * + * @param bufferSource Native stream input chunk. + * @returns Copied bytes for the mock stream output. + */ +function copyBufferSource(bufferSource: BufferSource): Uint8Array { + if (bufferSource instanceof ArrayBuffer) { + return new Uint8Array(bufferSource).slice(); + } + return new Uint8Array( + bufferSource.buffer, + bufferSource.byteOffset, + bufferSource.byteLength + ).slice(); +} + /** Builds packed 24-bit fixed-point position fixture bytes. */ function makePositionStream(): Uint8Array { const positions = new Uint8Array(18); diff --git a/scripts/test.mjs b/scripts/test.mjs index fd6d0f5770..2133e8bcc5 100644 --- a/scripts/test.mjs +++ b/scripts/test.mjs @@ -49,7 +49,7 @@ const modeArguments = { if (mode === 'bench') { const benchLoaderRegistration = - 'data:text/javascript,import { register } from "node:module"; import { pathToFileURL } from "node:url"; register("ts-node/esm", pathToFileURL("./")); register("@loaders.gl/devtools-extensions/bench-loader", pathToFileURL("./"));'; + 'data:text/javascript,import { register } from "node:module"; import { pathToFileURL } from "node:url"; register("@loaders.gl/devtools-extensions/bench-loader", pathToFileURL("./"));'; process.exitCode = await runProcess( 'node', ['--import', benchLoaderRegistration, './test/bench/node.js', ...passthroughArgs] diff --git a/tsconfig.json b/tsconfig.json index 38eac8b8c6..4699f8e82c 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -82,6 +82,9 @@ "@loaders.gl/compression": [ "modules/compression/src" ], + "@loaders.gl/compression/*": [ + "modules/compression/src/*" + ], "@loaders.gl/compression/test": [ "modules/compression/test" ], diff --git a/yarn.lock b/yarn.lock index 2a7df88543..815418ffdc 100644 --- a/yarn.lock +++ b/yarn.lock @@ -5283,15 +5283,6 @@ __metadata: zstd-codec: "npm:^0.1" peerDependencies: "@loaders.gl/core": ~5.0.0-alpha.0 - dependenciesMeta: - "@types/brotli": - optional: true - brotli: - optional: true - lz4js: - optional: true - zstd-codec: - optional: true languageName: unknown linkType: soft