From 0220f41ea6073acc24ec3eb673bf6177cb9d3d64 Mon Sep 17 00:00:00 2001 From: Ib Green Date: Fri, 10 Jul 2026 09:34:45 -0400 Subject: [PATCH 1/6] add native decompression stream support --- docs/modules/compression/README.md | 14 +- .../api-reference/brotli-compression.md | 7 + .../api-reference/deflate-compression.md | 6 + .../api-reference/gzip-compression.md | 5 + .../api-reference/zstd-compression.md | 13 + .../parquet/api-reference/parquet-loader.md | 6 +- .../splats/api-reference/spz-loader.md | 7 +- docs/whats-new.mdx | 4 + .../compression/src/lib/brotli-compression.ts | 17 +- modules/compression/src/lib/compression.ts | 45 ++++ .../src/lib/decompression-stream.ts | 136 ++++++++++ .../src/lib/deflate-compression.ts | 37 +++ .../compression/src/lib/zstd-compression.ts | 15 +- modules/compression/test/compression.spec.ts | 245 ++++++++++++++++++ .../test/decompression-stream.node.spec.ts | 54 ++++ 15 files changed, 599 insertions(+), 12 deletions(-) create mode 100644 modules/compression/src/lib/decompression-stream.ts create mode 100644 modules/compression/test/decompression-stream.node.spec.ts diff --git a/docs/modules/compression/README.md b/docs/modules/compression/README.md index 7b308f3a83..607ad95418 100644 --- a/docs/modules/compression/README.md +++ b/docs/modules/compression/README.md @@ -7,17 +7,23 @@ 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 +Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's +`DecompressionStream` implementation for gzip, deflate, raw deflate, Brotli, and Zstandard +when that exact format is supported. If the API or format is unavailable, loaders.gl falls back +to its existing codec implementation. Compression and synchronous decompression keep their +existing codec requirements. From-v5.0 + ## API | Compression Class | Format | Characteristics | Library Size | Notes | | ----------------------------------------------------------------------------------- | --------------------- | ------------------------------------ | ---------------------------------------------------- | ----- | | [`NoCompression`](/docs/modules/compression/api-reference/no-compression) | none | - | - | -| [`GzipCompression`](/docs/modules/compression/api-reference/gzip-compression) | gzip(`.gz`) | size | [Small](https://bundlephobia.com/package/pako) | -| [`DeflateCompression`](/docs/modules/compression/api-reference/deflate-compression) | DEFLATE(PKZIP) | size | [Small](https://bundlephobia.com/package/pako) | +| [`GzipCompression`](/docs/modules/compression/api-reference/gzip-compression) | gzip(`.gz`) | size | Native async decode; [small fallback](https://bundlephobia.com/package/pako) | +| [`DeflateCompression`](/docs/modules/compression/api-reference/deflate-compression) | DEFLATE(PKZIP) | size | Native async decode; [small fallback](https://bundlephobia.com/package/pako) | | [`LZ4Compression`](/docs/modules/compression/api-reference/lz4-compression) | LZ4 | speed ("real-time") | [Medium](https://bundlephobia.com/package/lz4) | -| [`ZstdCompression`](/docs/modules/compression/api-reference/zstd-compression) | Zstandard | speed ("real-time") | [Large](https://bundlephobia.com/package/zstd-codec) | +| [`ZstdCompression`](/docs/modules/compression/api-reference/zstd-compression) | Zstandard | speed ("real-time") | Native async decode when available; [large fallback](https://bundlephobia.com/package/zstd-codec) | | [`SnappyCompression`](/docs/modules/compression/api-reference/snappy-compression) | Snappy(Zippy) | speed ("real-time") | [Small](https://bundlephobia.com/package/snappys) | -| [`BrotliCompression`](/docs/modules/compression/api-reference/brotli-compression) | Brotli | Size, fast decompress, slow compress | [Large](https://bundlephobia.com/package/brotli) | +| [`BrotliCompression`](/docs/modules/compression/api-reference/brotli-compression) | Brotli | Size, fast decompress, slow compress | Native async decode when available; [large fallback](https://bundlephobia.com/package/brotli) | | [`LZOCompression`](/docs/modules/compression/api-reference/lzo-compression) | Lempel-Ziv-Oberheimer | size | Node.js only | ## Compression Formats diff --git a/docs/modules/compression/api-reference/brotli-compression.md b/docs/modules/compression/api-reference/brotli-compression.md index 34e41ea924..1890abfb4a 100644 --- a/docs/modules/compression/api-reference/brotli-compression.md +++ b/docs/modules/compression/api-reference/brotli-compression.md @@ -6,6 +6,11 @@ Compresses / decompresses Brotli encoded data. +Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's native +`DecompressionStream('brotli')` implementation when available, falling back to the existing codec +otherwise. Explicit codec options, compression, and synchronous decompression keep their existing +codec requirements. From-v5.0 + ## Interface Implements the [`Compression](./compression) API. @@ -13,3 +18,5 @@ Implements the [`Compression](./compression) API. ## Methods ### `constructor(options?: object)` + +`options` is optional for native asynchronous decompression. diff --git a/docs/modules/compression/api-reference/deflate-compression.md b/docs/modules/compression/api-reference/deflate-compression.md index 30282e4382..f2fadcc495 100644 --- a/docs/modules/compression/api-reference/deflate-compression.md +++ b/docs/modules/compression/api-reference/deflate-compression.md @@ -6,6 +6,12 @@ Compresses / decompresses DEFLATE encoded data. +Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's native +`DecompressionStream('deflate')` implementation when available. With `raw: true`, loaders.gl +probes `DecompressionStream('deflate-raw')`. It falls back to the existing codec when the native +format is unavailable; other explicit codec options, compression, and synchronous decompression keep +their existing codec requirements. From-v5.0 + ## Interface Implements the [`Compression](./compression) API. diff --git a/docs/modules/compression/api-reference/gzip-compression.md b/docs/modules/compression/api-reference/gzip-compression.md index 106be7e839..47dcf92149 100644 --- a/docs/modules/compression/api-reference/gzip-compression.md +++ b/docs/modules/compression/api-reference/gzip-compression.md @@ -6,6 +6,11 @@ Compresses / decompresses GZIP encoded data. +Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's native +`DecompressionStream('gzip')` implementation when available, falling back to the existing codec +otherwise. Explicit codec options, compression, and synchronous decompression keep their existing +codec requirements. From-v5.0 + ## Interface Implements the [`Compression](./compression) API. diff --git a/docs/modules/compression/api-reference/zstd-compression.md b/docs/modules/compression/api-reference/zstd-compression.md index 60a97efb5c..0d701e21bb 100644 --- a/docs/modules/compression/api-reference/zstd-compression.md +++ b/docs/modules/compression/api-reference/zstd-compression.md @@ -6,6 +6,15 @@ Compresses / decompresses Zstandard encoded data. +Asynchronous `decompress()` and `decompressBatches()` first use +`new DecompressionStream('zstd')` when the runtime accepts that format. In those runtimes, +applications do not need to install or inject `zstd-codec` for async decompression. +From-v5.0 + +When native Zstandard decompression is unavailable, inject `zstd-codec` through +`options.modules`. `compress()`, `compressSync()`, and `decompressSync()` continue to +require `zstd-codec` in every runtime. + ## Interface Implements the [`Compression](./compression) API. @@ -13,3 +22,7 @@ Implements the [`Compression](./compression) API. ## Methods ### `constructor(options?: object)` + +`options` is optional for native asynchronous decompression. Supply +`{modules: {'zstd-codec': ZstdCodec}}` when a fallback codec or synchronous/compression API is +needed. diff --git a/docs/modules/parquet/api-reference/parquet-loader.md b/docs/modules/parquet/api-reference/parquet-loader.md index 5c00766614..9e1853306f 100644 --- a/docs/modules/parquet/api-reference/parquet-loader.md +++ b/docs/modules/parquet/api-reference/parquet-loader.md @@ -119,8 +119,10 @@ field-level GeoArrow metadata. ## Compressions -Some compressions are big and need to be imported explicitly by the application -and passed to the `ParquetLoader` +Some compression codecs are big and need to be imported explicitly by the application and passed +to the `ParquetLoader`. Zstandard pages can be decompressed without `zstd-codec` when the +runtime supports `new DecompressionStream('zstd')`; inject `zstd-codec` as the fallback for +other runtimes. 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..edab6f96ad 100644 --- a/docs/modules/splats/api-reference/spz-loader.md +++ b/docs/modules/splats/api-reference/spz-loader.md @@ -17,9 +17,12 @@ ## 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. When the runtime supports +`new DecompressionStream('zstd')`, async SPZ parsing does not require an external ZSTD module. +Otherwise inject `zstd-codec` through loader options as the fallback. ```typescript +// Install zstd-codec only when the runtime lacks native zstd decompression. // npm install @loaders.gl/core @loaders.gl/splats zstd-codec import {load} from '@loaders.gl/core'; @@ -72,4 +75,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}` to decode SPZ version 4 streams when native zstd decompression is unavailable. | diff --git a/docs/whats-new.mdx b/docs/whats-new.mdx index beb0c98325..269661b8e5 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** + +- Async gzip, deflate, raw deflate, Brotli, and Zstandard decompression now probes and uses the runtime's native `DecompressionStream` before falling back to existing codecs. Runtimes with native Zstandard support no longer require `zstd-codec` for async decompression APIs, including Parquet and SPZ 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/src/lib/brotli-compression.ts b/modules/compression/src/lib/brotli-compression.ts index 3e22af80d5..e9131cf379 100644 --- a/modules/compression/src/lib/brotli-compression.ts +++ b/modules/compression/src/lib/brotli-compression.ts @@ -48,7 +48,17 @@ export class BrotliCompression extends Compression { readonly isSupported = true; readonly options: BrotliCompressionOptions; - constructor(options: BrotliCompressionOptions) { + /** Native Brotli format used for default asynchronous decompression. */ + protected get decompressionStreamFormat(): 'brotli' { + return 'brotli'; + } + + /** Only use native decompression when no codec-specific options would be ignored. */ + protected get useNativeDecompressionStream(): boolean { + return !this.options.brotli || Object.keys(this.options.brotli).length === 0; + } + + constructor(options: BrotliCompressionOptions = {}) { super(options); this.options = options; registerJSModules(options?.modules); @@ -90,6 +100,11 @@ export class BrotliCompression extends Compression { } async decompress(input: ArrayBuffer): Promise { + const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); + if (nativeOutput) { + return nativeOutput; + } + // On Node.js we can use built-in zlib if (!isBrowser && this.options.brotli?.useZlib) { const buffer = await promisify1(zlib.brotliDecompress)(input); diff --git a/modules/compression/src/lib/compression.ts b/modules/compression/src/lib/compression.ts index 446bcf6093..1dc383ac48 100644 --- a/modules/compression/src/lib/compression.ts +++ b/modules/compression/src/lib/compression.ts @@ -4,6 +4,11 @@ // Compression interface import {concatenateArrayBuffersAsync, registerJSModules} from '@loaders.gl/loader-utils'; +import { + decompressBatchesWithNativeDecompressionStream, + decompressWithNativeDecompressionStream, + type NativeDecompressionFormat +} from './decompression-stream'; /** Compression options */ export type CompressionOptions = { @@ -18,6 +23,16 @@ export abstract class Compression { abstract readonly contentEncodings: string[]; abstract readonly isSupported: boolean; + /** Native format used for default asynchronous decompression, when available. */ + protected get decompressionStreamFormat(): NativeDecompressionFormat | undefined { + return undefined; + } + + /** Whether default asynchronous decompression can use the native stream path. */ + protected get useNativeDecompressionStream(): boolean { + return true; + } + constructor(options?: CompressionOptions) { this.compressBatches = this.compressBatches.bind(this); this.decompressBatches = this.decompressBatches.bind(this); @@ -37,6 +52,10 @@ export abstract class Compression { /** Asynchronously decompress data */ async decompress(input: ArrayBuffer, size?: number): Promise { + const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); + if (nativeOutput) { + return nativeOutput; + } await this.preload(); return this.decompressSync(input, size); } @@ -64,6 +83,17 @@ export abstract class Compression { async *decompressBatches( asyncIterator: AsyncIterable | Iterable ): AsyncIterable { + if (this.decompressionStreamFormat && this.useNativeDecompressionStream) { + const outputBatches = decompressBatchesWithNativeDecompressionStream( + asyncIterator, + this.decompressionStreamFormat + ); + if (outputBatches) { + yield* outputBatches; + return; + } + } + // TODO - implement incremental compression const input = await this.concatenate(asyncIterator); yield this.decompress(input); @@ -75,6 +105,21 @@ export abstract class Compression { return concatenateArrayBuffersAsync(asyncIterator); } + /** + * Attempts native asynchronous decompression for classes that declare a stream format. + * + * @param input Compressed input data. + * @returns Decompressed data, or null when native decompression should not be used. + */ + protected async tryDecompressWithNativeDecompressionStream( + input: ArrayBuffer + ): Promise { + if (!this.decompressionStreamFormat || !this.useNativeDecompressionStream) { + return null; + } + return await decompressWithNativeDecompressionStream(input, this.decompressionStreamFormat); + } + protected improveError(error) { if (!error.message.includes(this.name)) { error.message = `${this.name} ${error.message}`; diff --git a/modules/compression/src/lib/decompression-stream.ts b/modules/compression/src/lib/decompression-stream.ts new file mode 100644 index 0000000000..9847fe3626 --- /dev/null +++ b/modules/compression/src/lib/decompression-stream.ts @@ -0,0 +1,136 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import {concatenateArrayBuffersAsync, ensureArrayBuffer} from '@loaders.gl/loader-utils'; + +/** + * 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 concatenateArrayBuffersAsync(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 ensureArrayBuffer(value); + } + await writePromise; + } finally { + if (!outputCompleted) { + await reader.cancel().catch(() => {}); + await writer.abort().catch(() => {}); + await writePromise.catch(() => {}); + } + reader.releaseLock(); + } +} + +/** + * 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/deflate-compression.ts b/modules/compression/src/lib/deflate-compression.ts index e6345ee234..a494334b67 100644 --- a/modules/compression/src/lib/deflate-compression.ts +++ b/modules/compression/src/lib/deflate-compression.ts @@ -5,6 +5,10 @@ // DEFLATE import type {CompressionOptions} from './compression'; import {Compression} from './compression'; +import { + decompressBatchesWithNativeDecompressionStream, + type NativeDecompressionFormat +} from './decompression-stream'; import {isBrowser, toArrayBuffer, promisify1} from '@loaders.gl/loader-utils'; import pako from 'pako'; // https://bundlephobia.com/package/pako import zlib from 'zlib'; @@ -26,6 +30,23 @@ export class DeflateCompression extends Compression { readonly options: DeflateCompressionOptions; + /** Native format matching the configured wrapper type. */ + protected get decompressionStreamFormat(): NativeDecompressionFormat { + if (this.options.raw) { + return 'deflate-raw'; + } + return this.options.deflate?.gzip ? 'gzip' : 'deflate'; + } + + /** Only use native decompression when no codec-specific options would be ignored. */ + protected get useNativeDecompressionStream(): boolean { + const deflateOptions = this.options.deflate; + if (!deflateOptions) { + return true; + } + return Object.keys(deflateOptions).every(optionName => optionName === 'gzip'); + } + private _chunks: ArrayBuffer[] = []; constructor(options: DeflateCompressionOptions = {}) { @@ -45,6 +66,11 @@ export class DeflateCompression extends Compression { } async decompress(input: ArrayBuffer): Promise { + const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); + if (nativeOutput) { + return nativeOutput; + } + // On Node.js we can use built-in zlib if (!isBrowser && this.options.deflate?.useZlib) { const buffer = this.options.deflate?.gzip @@ -90,6 +116,17 @@ export class DeflateCompression extends Compression { async *decompressBatches( asyncIterator: AsyncIterable | Iterable ): AsyncIterable { + if (this.decompressionStreamFormat && this.useNativeDecompressionStream) { + const outputBatches = decompressBatchesWithNativeDecompressionStream( + asyncIterator, + this.decompressionStreamFormat + ); + if (outputBatches) { + yield* outputBatches; + return; + } + } + const pakoOptions: pako.InflateOptions = this.options?.deflate || {}; const pakoProcessor = new pako.Inflate(pakoOptions); yield* this.transformBatches(pakoProcessor, asyncIterator); diff --git a/modules/compression/src/lib/zstd-compression.ts b/modules/compression/src/lib/zstd-compression.ts index daa876937a..37734edd4e 100644 --- a/modules/compression/src/lib/zstd-compression.ts +++ b/modules/compression/src/lib/zstd-compression.ts @@ -7,7 +7,6 @@ import type {CompressionOptions} from './compression'; import {Compression} from './compression'; import { registerJSModules, - checkJSModule, getJSModule, getJSModuleOrNull, ensureArrayBuffer @@ -30,11 +29,16 @@ export class ZstdCompression extends Compression { readonly isSupported = true; readonly options: CompressionOptions; + /** Native Zstandard format used for default asynchronous decompression. */ + protected get decompressionStreamFormat(): 'zstd' { + return 'zstd'; + } + /** * 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); @@ -42,7 +46,6 @@ export class ZstdCompression extends Compression { async preload(modules: Record = {}): Promise { registerJSModules(modules); - checkJSModule('zstd-codec', this.name); const ZstdCodec = getJSModuleOrNull('zstd-codec'); // eslint-disable-next-line @typescript-eslint/no-misused-promises if (!zstdPromise && ZstdCodec) { @@ -70,7 +73,13 @@ export class ZstdCompression extends Compression { } async decompress(input: ArrayBuffer, size?: number): Promise { + const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); + if (nativeOutput) { + return nativeOutput; + } + await this.preload(); + getJSModule('zstd-codec', this.name); const simpleZstd = new zstd.Streaming(); const inputArray = new Uint8Array(input); diff --git a/modules/compression/test/compression.spec.ts b/modules/compression/test/compression.spec.ts index 88e97bb266..075cb05256 100644 --- a/modules/compression/test/compression.spec.ts +++ b/modules/compression/test/compression.spec.ts @@ -162,6 +162,155 @@ test('compression#batched', async t => { t.end(); }); +test('gzip#native DecompressionStream atomic and batched', async t => { + if (typeof globalThis.DecompressionStream === 'undefined') { + t.comment('DecompressionStream is not available in this runtime'); + t.end(); + return; + } + + try { + const probeStream = new globalThis.DecompressionStream('gzip'); + await probeStream.writable.abort(); + } catch { + t.comment('gzip DecompressionStream is not available in this runtime'); + t.end(); + return; + } + + const inputData = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]).buffer; + const compression = new GZipCompression(); + const compressedData = compression.compressSync(inputData); + + const decompressedData = await compression.decompress(compressedData); + t.ok(compareArrayBuffers(inputData, decompressedData), 'native atomic gzip decompression works'); + + const compressedBatches = [ + compressedData.slice(0, 5), + compressedData.slice(5, compressedData.byteLength) + ]; + const decompressedBatches = compression.decompressBatches(compressedBatches); + const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); + t.ok( + compareArrayBuffers(inputData, decompressedBatchData), + 'native batched gzip decompression works' + ); + t.end(); +}); + +test('zstd#native DecompressionStream works without zstd-codec', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream({ + formats, + supportedFormats: ['zstd'] + }); + const restoreZstdCodec = removeRegisteredModule('zstd-codec'); + + try { + const inputBatches = [new Uint8Array([1, 2, 3]).buffer, new Uint8Array([4, 5, 6]).buffer]; + const inputData = concatenateArrayBuffers(...inputBatches); + const compression = new ZstdCompression(); + + const decompressedData = await compression.decompress(inputData); + t.ok(compareArrayBuffers(inputData, decompressedData), 'native atomic zstd needs no codec'); + + const decompressedBatches = compression.decompressBatches(inputBatches); + 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 { + restoreZstdCodec(); + restoreDecompressionStream(); + } + + t.end(); +}); + +test('zstd#unsupported native format falls back to zstd-codec', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream({ + formats, + supportedFormats: [] + }); + + try { + const inputData = new Uint8Array([1, 2, 3, 4, 5, 6]).buffer; + const compression = new ZstdCompression({modules}); + const compressedData = await compression.compress(inputData); + const decompressedData = await compression.decompress(compressedData); + + t.ok(compareArrayBuffers(inputData, decompressedData), 'zstd codec fallback decompresses data'); + t.deepEqual(formats, ['zstd'], 'zstd native support is probed before falling back'); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + +test('zstd#native stream failures do not fall back', async t => { + const restoreDecompressionStream = installMockDecompressionStream({ + formats: [], + supportedFormats: ['zstd'], + failWith: new Error('mock native decompression failed') + }); + + try { + const inputData = new Uint8Array([1, 2, 3]).buffer; + const compression = new ZstdCompression({modules}); + await t.rejects( + compression.decompress(inputData), + /mock native decompression failed/, + 'native stream errors propagate' + ); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + +test('deflate#native format mapping and explicit option fallback', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream({ + formats, + supportedFormats: ['deflate-raw'] + }); + + try { + const inputData = new Uint8Array([1, 2, 3, 4, 5, 6]).buffer; + const rawCompression = new DeflateCompression({raw: true}); + const decompressedRawData = await rawCompression.decompress(inputData); + + t.ok(compareArrayBuffers(inputData, decompressedRawData), 'raw deflate uses native stream'); + t.deepEqual(formats, ['deflate-raw'], 'raw deflate maps to deflate-raw'); + + formats.length = 0; + const compressedData = new DeflateCompression().compressSync(inputData); + const configuredCompression = new DeflateCompression({deflate: {useZlib: true}}); + const decompressedConfiguredData = await configuredCompression.decompress(compressedData); + + t.ok( + compareArrayBuffers(inputData, decompressedConfiguredData), + 'configured deflate uses the existing implementation' + ); + t.deepEqual(formats, [], 'codec-specific options bypass the native stream'); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + +test('compression#native constructors accept omitted options', t => { + t.ok(new BrotliCompression(), 'BrotliCompression options are optional'); + t.ok(new ZstdCompression(), 'ZstdCompression options are optional'); + t.end(); +}); + // WORKER TESTS test('gzip#worker', async t => { const {binaryData} = getData(); @@ -260,3 +409,99 @@ 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; + } + }; +} + +/** + * 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(); +} 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..0717ecc9fe --- /dev/null +++ b/modules/compression/test/decompression-stream.node.spec.ts @@ -0,0 +1,54 @@ +// loaders.gl +// SPDX-License-Identifier: MIT +// Copyright (c) vis.gl contributors + +import test from 'tape-promise/tape'; +import {GZipCompression} from '@loaders.gl/compression'; +import {compareArrayBuffers} from './utils/test-utils'; + +type MutableGlobalThis = typeof globalThis & { + Buffer?: typeof Buffer; +}; + +test('gzip#native DecompressionStream accepts ArrayBuffer input in Node.js', async t => { + if (typeof globalThis.DecompressionStream === 'undefined') { + t.comment('DecompressionStream is not available in this runtime'); + t.end(); + return; + } + + try { + const probeStream = new globalThis.DecompressionStream('gzip'); + await probeStream.writable.abort(); + } catch { + t.comment('gzip DecompressionStream is not available in this runtime'); + t.end(); + return; + } + + const inputData = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]).buffer; + const compression = new GZipCompression(); + const compressedData = compression.compressSync(inputData); + const decompressedData = await compression.decompress(compressedData); + + t.ok(compareArrayBuffers(inputData, decompressedData), 'native gzip accepts ArrayBuffer input'); + t.end(); +}); + +test('gzip#native DecompressionStream falls back without global Buffer in Node.js', async t => { + const inputData = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]).buffer; + const compression = new GZipCompression(); + const compressedData = compression.compressSync(inputData); + const mutableGlobalThis = globalThis as MutableGlobalThis; + const originalBuffer = mutableGlobalThis.Buffer; + mutableGlobalThis.Buffer = undefined; + + try { + const decompressedData = await compression.decompress(compressedData); + t.ok(compareArrayBuffers(inputData, decompressedData), 'gzip falls back without Buffer'); + } finally { + mutableGlobalThis.Buffer = originalBuffer; + } + + t.end(); +}); From ce9471df57387284bd936403854f735097abb4d8 Mon Sep 17 00:00:00 2001 From: Ib Green Date: Fri, 10 Jul 2026 11:10:54 -0400 Subject: [PATCH 2/6] prefer readonly compression fields --- AGENTS.md | 1 + .../compression/src/lib/brotli-compression.ts | 9 +++----- modules/compression/src/lib/compression.ts | 8 ++----- .../src/lib/deflate-compression.ts | 23 ++++++++----------- .../compression/src/lib/zstd-compression.ts | 4 +--- 5 files changed, 17 insertions(+), 28 deletions(-) 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/modules/compression/src/lib/brotli-compression.ts b/modules/compression/src/lib/brotli-compression.ts index e9131cf379..0839962360 100644 --- a/modules/compression/src/lib/brotli-compression.ts +++ b/modules/compression/src/lib/brotli-compression.ts @@ -49,18 +49,15 @@ export class BrotliCompression extends Compression { readonly options: BrotliCompressionOptions; /** Native Brotli format used for default asynchronous decompression. */ - protected get decompressionStreamFormat(): 'brotli' { - return 'brotli'; - } + protected readonly decompressionStreamFormat = 'brotli'; /** Only use native decompression when no codec-specific options would be ignored. */ - protected get useNativeDecompressionStream(): boolean { - return !this.options.brotli || Object.keys(this.options.brotli).length === 0; - } + protected readonly useNativeDecompressionStream: boolean; constructor(options: BrotliCompressionOptions = {}) { super(options); this.options = options; + this.useNativeDecompressionStream = !options.brotli || Object.keys(options.brotli).length === 0; registerJSModules(options?.modules); } diff --git a/modules/compression/src/lib/compression.ts b/modules/compression/src/lib/compression.ts index 1dc383ac48..be70734bde 100644 --- a/modules/compression/src/lib/compression.ts +++ b/modules/compression/src/lib/compression.ts @@ -24,14 +24,10 @@ export abstract class Compression { abstract readonly isSupported: boolean; /** Native format used for default asynchronous decompression, when available. */ - protected get decompressionStreamFormat(): NativeDecompressionFormat | undefined { - return undefined; - } + protected readonly decompressionStreamFormat: NativeDecompressionFormat | undefined = undefined; /** Whether default asynchronous decompression can use the native stream path. */ - protected get useNativeDecompressionStream(): boolean { - return true; - } + protected readonly useNativeDecompressionStream: boolean = true; constructor(options?: CompressionOptions) { this.compressBatches = this.compressBatches.bind(this); diff --git a/modules/compression/src/lib/deflate-compression.ts b/modules/compression/src/lib/deflate-compression.ts index a494334b67..23c4c46600 100644 --- a/modules/compression/src/lib/deflate-compression.ts +++ b/modules/compression/src/lib/deflate-compression.ts @@ -31,27 +31,24 @@ export class DeflateCompression extends Compression { readonly options: DeflateCompressionOptions; /** Native format matching the configured wrapper type. */ - protected get decompressionStreamFormat(): NativeDecompressionFormat { - if (this.options.raw) { - return 'deflate-raw'; - } - return this.options.deflate?.gzip ? 'gzip' : 'deflate'; - } + protected readonly decompressionStreamFormat: NativeDecompressionFormat; /** Only use native decompression when no codec-specific options would be ignored. */ - protected get useNativeDecompressionStream(): boolean { - const deflateOptions = this.options.deflate; - if (!deflateOptions) { - return true; - } - return Object.keys(deflateOptions).every(optionName => optionName === 'gzip'); - } + protected readonly useNativeDecompressionStream: boolean; private _chunks: ArrayBuffer[] = []; constructor(options: DeflateCompressionOptions = {}) { super(options); this.options = options; + this.decompressionStreamFormat = options.raw + ? 'deflate-raw' + : options.deflate?.gzip + ? 'gzip' + : 'deflate'; + const deflateOptions = options.deflate; + this.useNativeDecompressionStream = + !deflateOptions || Object.keys(deflateOptions).every(optionName => optionName === 'gzip'); } async compress(input: ArrayBuffer): Promise { diff --git a/modules/compression/src/lib/zstd-compression.ts b/modules/compression/src/lib/zstd-compression.ts index 37734edd4e..7d35620e4a 100644 --- a/modules/compression/src/lib/zstd-compression.ts +++ b/modules/compression/src/lib/zstd-compression.ts @@ -30,9 +30,7 @@ export class ZstdCompression extends Compression { readonly options: CompressionOptions; /** Native Zstandard format used for default asynchronous decompression. */ - protected get decompressionStreamFormat(): 'zstd' { - return 'zstd'; - } + protected readonly decompressionStreamFormat = 'zstd'; /** * zstd-codec is an injectable dependency due to big size From e6d6d85be9f40be4dffe36ece6621e1b92988aff Mon Sep 17 00:00:00 2001 From: Ib Green Date: Tue, 14 Jul 2026 05:42:09 -0400 Subject: [PATCH 3/6] clarify future native zstd support --- docs/whats-new.mdx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/whats-new.mdx b/docs/whats-new.mdx index 269661b8e5..0e180ab307 100644 --- a/docs/whats-new.mdx +++ b/docs/whats-new.mdx @@ -75,7 +75,7 @@ Release Date: 2026 **@loaders.gl/compression** -- Async gzip, deflate, raw deflate, Brotli, and Zstandard decompression now probes and uses the runtime's native `DecompressionStream` before falling back to existing codecs. Runtimes with native Zstandard support no longer require `zstd-codec` for async decompression APIs, including Parquet and SPZ parsing. +- Async gzip, deflate, raw deflate, and Brotli decompression now probes and uses the runtime's native `DecompressionStream` before falling back to existing codecs. Zstandard is also probed so future runtimes can use native support automatically; until native Zstandard support becomes widely available, most runtimes still need `zstd-codec` for Zstandard-backed Parquet and SPZ parsing. **@loaders.gl/deck-layers** From 007f7083ca896ccb824e8fd3dfc5c906c886efec Mon Sep 17 00:00:00 2001 From: Ib Green Date: Sat, 1 Aug 2026 07:59:38 -0400 Subject: [PATCH 4/6] prefer provided decompression codecs --- docs/modules/compression/README.md | 11 +- .../api-reference/brotli-compression.md | 5 +- .../api-reference/zstd-compression.md | 18 +-- .../parquet/api-reference/parquet-loader.md | 7 +- .../splats/api-reference/spz-loader.md | 11 +- .../compression/src/lib/brotli-compression.ts | 3 + modules/compression/src/lib/compression.ts | 26 +++- .../compression/src/lib/zstd-compression.ts | 3 + modules/compression/test/compression.spec.ts | 147 ++++++++++++++---- .../test/decompression-stream.node.spec.ts | 83 +++++++--- .../utils/native-decompression-test-utils.ts | 82 ++++++++++ 11 files changed, 315 insertions(+), 81 deletions(-) create mode 100644 modules/compression/test/utils/native-decompression-test-utils.ts diff --git a/docs/modules/compression/README.md b/docs/modules/compression/README.md index 607ad95418..4a9992d1c2 100644 --- a/docs/modules/compression/README.md +++ b/docs/modules/compression/README.md @@ -7,11 +7,12 @@ 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 -Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's -`DecompressionStream` implementation for gzip, deflate, raw deflate, Brotli, and Zstandard -when that exact format is supported. If the API or format is unavailable, loaders.gl falls back -to its existing codec implementation. Compression and synchronous decompression keep their -existing codec requirements. From-v5.0 +When no relevant codec module is registered, default asynchronous `decompress()` and +`decompressBatches()` calls use the runtime's `DecompressionStream` implementation for gzip, +deflate, raw deflate, Brotli, and Zstandard when that exact format is supported. If a codec module +is provided, loaders.gl uses it instead of silently switching implementations. Compression and +synchronous decompression keep their existing codec requirements. +From-v5.0 ## API diff --git a/docs/modules/compression/api-reference/brotli-compression.md b/docs/modules/compression/api-reference/brotli-compression.md index 1890abfb4a..d709df668a 100644 --- a/docs/modules/compression/api-reference/brotli-compression.md +++ b/docs/modules/compression/api-reference/brotli-compression.md @@ -8,8 +8,9 @@ Compresses / decompresses Brotli encoded data. Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's native `DecompressionStream('brotli')` implementation when available, falling back to the existing codec -otherwise. Explicit codec options, compression, and synchronous decompression keep their existing -codec requirements. From-v5.0 +otherwise. A provided `brotli` module takes precedence over the native stream path. Explicit codec +options, compression, and synchronous decompression keep their existing codec requirements. +From-v5.0 ## Interface diff --git a/docs/modules/compression/api-reference/zstd-compression.md b/docs/modules/compression/api-reference/zstd-compression.md index 0d701e21bb..32242a7775 100644 --- a/docs/modules/compression/api-reference/zstd-compression.md +++ b/docs/modules/compression/api-reference/zstd-compression.md @@ -6,14 +6,14 @@ Compresses / decompresses Zstandard encoded data. -Asynchronous `decompress()` and `decompressBatches()` first use -`new DecompressionStream('zstd')` when the runtime accepts that format. In those runtimes, -applications do not need to install or inject `zstd-codec` for async decompression. +When no `zstd-codec` module is registered, asynchronous `decompress()` and +`decompressBatches()` probe `new DecompressionStream('zstd')` so future runtimes can use native +Zstandard support automatically. Native Zstandard support is not yet widely available. From-v5.0 -When native Zstandard decompression is unavailable, inject `zstd-codec` through -`options.modules`. `compress()`, `compressSync()`, and `decompressSync()` continue to -require `zstd-codec` in every runtime. +Inject `zstd-codec` through `options.modules` for broad compatibility. When it is provided, it +takes precedence over the native stream path. `compress()`, `compressSync()`, and +`decompressSync()` continue to require `zstd-codec` in every runtime. ## Interface @@ -23,6 +23,6 @@ Implements the [`Compression](./compression) API. ### `constructor(options?: object)` -`options` is optional for native asynchronous decompression. Supply -`{modules: {'zstd-codec': ZstdCodec}}` when a fallback codec or synchronous/compression API is -needed. +`options` is optional for future native asynchronous decompression. Supply +`{modules: {'zstd-codec': ZstdCodec}}` for broad runtime compatibility or when a +synchronous/compression API is needed. diff --git a/docs/modules/parquet/api-reference/parquet-loader.md b/docs/modules/parquet/api-reference/parquet-loader.md index 9e1853306f..57747da929 100644 --- a/docs/modules/parquet/api-reference/parquet-loader.md +++ b/docs/modules/parquet/api-reference/parquet-loader.md @@ -120,9 +120,10 @@ field-level GeoArrow metadata. ## Compressions Some compression codecs are big and need to be imported explicitly by the application and passed -to the `ParquetLoader`. Zstandard pages can be decompressed without `zstd-codec` when the -runtime supports `new DecompressionStream('zstd')`; inject `zstd-codec` as the fallback for -other runtimes. LZ4 still requires `lz4js`. From-v5.0 +to the `ParquetLoader`. Without an injected codec, Zstandard pages can use +`new DecompressionStream('zstd')` when a future runtime supports it. Inject `zstd-codec` for broad +runtime compatibility; when provided, loaders.gl uses it instead of the native stream 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 edab6f96ad..a84ef39b0c 100644 --- a/docs/modules/splats/api-reference/spz-loader.md +++ b/docs/modules/splats/api-reference/spz-loader.md @@ -17,12 +17,13 @@ ## Usage -SPZ version 4 uses ZSTD-compressed attribute streams. When the runtime supports -`new DecompressionStream('zstd')`, async SPZ parsing does not require an external ZSTD module. -Otherwise inject `zstd-codec` through loader options as the fallback. +SPZ version 4 uses ZSTD-compressed attribute streams. Without an injected codec, async SPZ parsing +can use `new DecompressionStream('zstd')` when a future runtime supports it. Inject `zstd-codec` +through loader options for broad runtime compatibility; when provided, loaders.gl uses it instead +of the native stream path. ```typescript -// Install zstd-codec only when the runtime lacks native zstd decompression. +// Install zstd-codec for broad runtime compatibility. // npm install @loaders.gl/core @loaders.gl/splats zstd-codec import {load} from '@loaders.gl/core'; @@ -75,4 +76,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` | `{}` | Include `{'zstd-codec': ZstdCodec}` to decode SPZ version 4 streams when native zstd decompression is unavailable. | +| `modules` | `object` | `{}` | Include `{'zstd-codec': ZstdCodec}` for broad SPZ version 4 runtime compatibility. | diff --git a/modules/compression/src/lib/brotli-compression.ts b/modules/compression/src/lib/brotli-compression.ts index 0839962360..20dd76fecd 100644 --- a/modules/compression/src/lib/brotli-compression.ts +++ b/modules/compression/src/lib/brotli-compression.ts @@ -54,6 +54,9 @@ export class BrotliCompression extends Compression { /** Only use native decompression when no codec-specific options would be ignored. */ protected readonly useNativeDecompressionStream: boolean; + /** Registered Brotli module that takes precedence over native decompression. */ + protected readonly decompressionModuleName = 'brotli'; + constructor(options: BrotliCompressionOptions = {}) { super(options); this.options = options; diff --git a/modules/compression/src/lib/compression.ts b/modules/compression/src/lib/compression.ts index be70734bde..977cc7a674 100644 --- a/modules/compression/src/lib/compression.ts +++ b/modules/compression/src/lib/compression.ts @@ -3,7 +3,11 @@ // Copyright (c) vis.gl contributors // Compression interface -import {concatenateArrayBuffersAsync, registerJSModules} from '@loaders.gl/loader-utils'; +import { + concatenateArrayBuffersAsync, + getJSModuleOrNull, + registerJSModules +} from '@loaders.gl/loader-utils'; import { decompressBatchesWithNativeDecompressionStream, decompressWithNativeDecompressionStream, @@ -29,6 +33,9 @@ export abstract class Compression { /** Whether default asynchronous decompression can use the native stream path. */ protected readonly useNativeDecompressionStream: boolean = true; + /** Registered fallback module that takes precedence over native decompression. */ + protected readonly decompressionModuleName: string | undefined = undefined; + constructor(options?: CompressionOptions) { this.compressBatches = this.compressBatches.bind(this); this.decompressBatches = this.decompressBatches.bind(this); @@ -79,7 +86,7 @@ export abstract class Compression { async *decompressBatches( asyncIterator: AsyncIterable | Iterable ): AsyncIterable { - if (this.decompressionStreamFormat && this.useNativeDecompressionStream) { + if (this.decompressionStreamFormat && this.shouldUseNativeDecompressionStream()) { const outputBatches = decompressBatchesWithNativeDecompressionStream( asyncIterator, this.decompressionStreamFormat @@ -110,12 +117,25 @@ export abstract class Compression { protected async tryDecompressWithNativeDecompressionStream( input: ArrayBuffer ): Promise { - if (!this.decompressionStreamFormat || !this.useNativeDecompressionStream) { + if (!this.decompressionStreamFormat || !this.shouldUseNativeDecompressionStream()) { return null; } return await decompressWithNativeDecompressionStream(input, this.decompressionStreamFormat); } + /** + * Returns whether native asynchronous decompression should be attempted. + * + * Explicitly registered fallback modules take precedence so applications that provide a + * library do not silently switch implementations when a runtime adds native support. + */ + protected shouldUseNativeDecompressionStream(): boolean { + return ( + this.useNativeDecompressionStream && + (!this.decompressionModuleName || !getJSModuleOrNull(this.decompressionModuleName)) + ); + } + protected improveError(error) { if (!error.message.includes(this.name)) { error.message = `${this.name} ${error.message}`; diff --git a/modules/compression/src/lib/zstd-compression.ts b/modules/compression/src/lib/zstd-compression.ts index 7d35620e4a..7baea362ae 100644 --- a/modules/compression/src/lib/zstd-compression.ts +++ b/modules/compression/src/lib/zstd-compression.ts @@ -32,6 +32,9 @@ export class ZstdCompression extends Compression { /** Native Zstandard format used for default asynchronous decompression. */ protected readonly decompressionStreamFormat = 'zstd'; + /** Registered Zstandard module that takes precedence over native decompression. */ + protected readonly decompressionModuleName = 'zstd-codec'; + /** * zstd-codec is an injectable dependency due to big size * @param options diff --git a/modules/compression/test/compression.spec.ts b/modules/compression/test/compression.spec.ts index 075cb05256..0431f0804d 100644 --- a/modules/compression/test/compression.spec.ts +++ b/modules/compression/test/compression.spec.ts @@ -18,6 +18,13 @@ import { 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,39 +169,65 @@ test('compression#batched', async t => { t.end(); }); -test('gzip#native DecompressionStream atomic and batched', async t => { - if (typeof globalThis.DecompressionStream === 'undefined') { - t.comment('DecompressionStream is not available in this runtime'); - t.end(); - return; - } - - try { - const probeStream = new globalThis.DecompressionStream('gzip'); - await probeStream.writable.abort(); - } catch { - t.comment('gzip DecompressionStream is not available in this runtime'); - t.end(); - return; - } +test('compression#native 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 inputData = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]).buffer; - const compression = new GZipCompression(); - const compressedData = compression.compressSync(inputData); + const restoreModule = + format === 'brotli' + ? removeRegisteredModule('brotli') + : format === 'zstd' + ? removeRegisteredModule('zstd-codec') + : null; + const nativeFormats: NativeDecompressionTestFormat[] = []; + const restoreDecompressionStream = installRecordingDecompressionStream(nativeFormats); + + try { + const compression = + format === 'gzip' + ? new GZipCompression() + : format === 'deflate' + ? new DeflateCompression() + : format === 'deflate-raw' + ? new DeflateCompression({raw: true}) + : format === 'brotli' + ? new BrotliCompression() + : new ZstdCompression(); + const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES[format]).buffer; + + const decompressedData = await compression.decompress(compressedData); + t.ok( + compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), + `native atomic ${format} decompression works` + ); - const decompressedData = await compression.decompress(compressedData); - t.ok(compareArrayBuffers(inputData, decompressedData), 'native atomic gzip 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 = compression.decompressBatches(compressedBatches); + 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(); + restoreModule?.(); + } + } - const compressedBatches = [ - compressedData.slice(0, 5), - compressedData.slice(5, compressedData.byteLength) - ]; - const decompressedBatches = compression.decompressBatches(compressedBatches); - const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); - t.ok( - compareArrayBuffers(inputData, decompressedBatchData), - 'native batched gzip decompression works' - ); t.end(); }); @@ -229,11 +262,11 @@ test('zstd#native DecompressionStream works without zstd-codec', async t => { t.end(); }); -test('zstd#unsupported native format falls back to zstd-codec', async t => { +test('zstd#provided zstd-codec bypasses native DecompressionStream', async t => { const formats: string[] = []; const restoreDecompressionStream = installMockDecompressionStream({ formats, - supportedFormats: [] + supportedFormats: ['zstd'] }); try { @@ -243,7 +276,51 @@ test('zstd#unsupported native format falls back to zstd-codec', async t => { const decompressedData = await compression.decompress(compressedData); t.ok(compareArrayBuffers(inputData, decompressedData), 'zstd codec fallback decompresses data'); - t.deepEqual(formats, ['zstd'], 'zstd native support is probed before falling back'); + const splitIndex = Math.max(1, Math.floor(compressedData.byteLength / 2)); + const decompressedBatches = compression.decompressBatches([ + compressedData.slice(0, splitIndex), + compressedData.slice(splitIndex, compressedData.byteLength) + ]); + const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); + t.ok( + compareArrayBuffers(inputData, decompressedBatchData), + 'zstd codec fallback decompresses batches' + ); + t.deepEqual(formats, [], 'provided zstd codec bypasses the native stream'); + } finally { + restoreDecompressionStream(); + } + + t.end(); +}); + +test('brotli#provided module bypasses native DecompressionStream', async t => { + const formats: string[] = []; + const restoreDecompressionStream = installMockDecompressionStream({ + formats, + supportedFormats: ['brotli'] + }); + + try { + const compression = new BrotliCompression({modules}); + const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES.brotli).buffer; + const decompressedData = await compression.decompress(compressedData); + + t.ok( + compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), + 'provided brotli module decompresses data' + ); + const splitIndex = Math.max(1, Math.floor(compressedData.byteLength / 2)); + const decompressedBatches = compression.decompressBatches([ + compressedData.slice(0, splitIndex), + compressedData.slice(splitIndex, compressedData.byteLength) + ]); + const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); + t.ok( + compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedBatchData), + 'provided brotli module decompresses batches' + ); + t.deepEqual(formats, [], 'provided brotli module bypasses the native stream'); } finally { restoreDecompressionStream(); } @@ -257,16 +334,18 @@ test('zstd#native stream failures do not fall back', async t => { supportedFormats: ['zstd'], failWith: new Error('mock native decompression failed') }); + const restoreZstdCodec = removeRegisteredModule('zstd-codec'); try { const inputData = new Uint8Array([1, 2, 3]).buffer; - const compression = new ZstdCompression({modules}); + const compression = new ZstdCompression(); await t.rejects( compression.decompress(inputData), /mock native decompression failed/, 'native stream errors propagate' ); } finally { + restoreZstdCodec(); restoreDecompressionStream(); } diff --git a/modules/compression/test/decompression-stream.node.spec.ts b/modules/compression/test/decompression-stream.node.spec.ts index 0717ecc9fe..67eb782f00 100644 --- a/modules/compression/test/decompression-stream.node.spec.ts +++ b/modules/compression/test/decompression-stream.node.spec.ts @@ -3,35 +3,78 @@ // Copyright (c) vis.gl contributors import test from 'tape-promise/tape'; -import {GZipCompression} from '@loaders.gl/compression'; +import { + BrotliCompression, + DeflateCompression, + GZipCompression, + ZstdCompression +} from '@loaders.gl/compression'; +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('gzip#native DecompressionStream accepts ArrayBuffer input in Node.js', async t => { - if (typeof globalThis.DecompressionStream === 'undefined') { - t.comment('DecompressionStream is not available in this runtime'); - t.end(); - return; - } +test('compression#native 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; + } - try { - const probeStream = new globalThis.DecompressionStream('gzip'); - await probeStream.writable.abort(); - } catch { - t.comment('gzip DecompressionStream is not available in this runtime'); - t.end(); - return; - } + const nativeFormats: NativeDecompressionTestFormat[] = []; + const restoreDecompressionStream = installRecordingDecompressionStream(nativeFormats); - const inputData = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]).buffer; - const compression = new GZipCompression(); - const compressedData = compression.compressSync(inputData); - const decompressedData = await compression.decompress(compressedData); + try { + const compression = + format === 'gzip' + ? new GZipCompression() + : format === 'deflate' + ? new DeflateCompression() + : format === 'deflate-raw' + ? new DeflateCompression({raw: true}) + : format === 'brotli' + ? new BrotliCompression() + : new ZstdCompression(); + const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES[format]).buffer; + + const decompressedData = await compression.decompress(compressedData); + t.ok( + 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 = compression.decompressBatches(compressedBatches); + 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.ok(compareArrayBuffers(inputData, decompressedData), 'native gzip accepts ArrayBuffer input'); 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; + } + }; +} From b65e146c96c81e09d7c34fc474df52b138ec2cd6 Mon Sep 17 00:00:00 2001 From: Ib Green Date: Sun, 2 Aug 2026 15:39:08 -0400 Subject: [PATCH 5/6] make native decompression lightweight --- docs/modules/compression/README.md | 37 +++- .../api-reference/brotli-compression.md | 8 +- .../api-reference/deflate-compression.md | 6 - .../api-reference/gzip-compression.md | 5 - .../api-reference/native-decompression.md | 38 ++++ .../api-reference/zstd-compression.md | 17 +- .../parquet/api-reference/parquet-loader.md | 11 +- .../splats/api-reference/spz-loader.md | 10 +- docs/whats-new.mdx | 2 +- modules/compression/package.json | 39 +++- modules/compression/src/brotli-compression.ts | 6 + .../compression/src/deflate-compression.ts | 6 + modules/compression/src/gzip-compression.ts | 6 + .../compression/src/lib/brotli-compression.ts | 15 -- modules/compression/src/lib/compression.ts | 63 +----- .../src/lib/decompression-stream.ts | 42 +++- .../src/lib/deflate-compression.ts | 34 --- .../compression/src/lib/zstd-compression.ts | 14 +- modules/compression/src/lz4-compression.ts | 5 + .../compression/src/native-decompression.ts | 9 + modules/compression/src/no-compression.ts | 5 + modules/compression/src/snappy-compression.ts | 5 + modules/compression/src/zstd-compression.ts | 5 + modules/compression/test/compression.spec.ts | 187 ++++------------- .../test/decompression-stream.node.spec.ts | 43 ++-- modules/parquet/package.json | 8 +- modules/parquet/src/parquetjs/compression.ts | 166 ++++++++++----- modules/parquet/test/init.ts | 3 +- .../parquet/test/parquet-compression.spec.ts | 193 ++++++++++++++++++ modules/splats/src/lib/parse-spz.ts | 60 +++++- modules/splats/test/spz-loader.spec.ts | 184 ++++++++++++++--- tsconfig.json | 3 + yarn.lock | 9 - 33 files changed, 791 insertions(+), 453 deletions(-) create mode 100644 docs/modules/compression/api-reference/native-decompression.md create mode 100644 modules/compression/src/brotli-compression.ts create mode 100644 modules/compression/src/deflate-compression.ts create mode 100644 modules/compression/src/gzip-compression.ts create mode 100644 modules/compression/src/lz4-compression.ts create mode 100644 modules/compression/src/native-decompression.ts create mode 100644 modules/compression/src/no-compression.ts create mode 100644 modules/compression/src/snappy-compression.ts create mode 100644 modules/compression/src/zstd-compression.ts create mode 100644 modules/parquet/test/parquet-compression.spec.ts diff --git a/docs/modules/compression/README.md b/docs/modules/compression/README.md index 4a9992d1c2..a66cadd7c5 100644 --- a/docs/modules/compression/README.md +++ b/docs/modules/compression/README.md @@ -7,24 +7,43 @@ 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 -When no relevant codec module is registered, default asynchronous `decompress()` and -`decompressBatches()` calls use the runtime's `DecompressionStream` implementation for gzip, -deflate, raw deflate, Brotli, and Zstandard when that exact format is supported. If a codec module -is provided, loaders.gl uses it instead of silently switching implementations. Compression and -synchronous decompression keep their existing codec requirements. +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 | | ----------------------------------------------------------------------------------- | --------------------- | ------------------------------------ | ---------------------------------------------------- | ----- | | [`NoCompression`](/docs/modules/compression/api-reference/no-compression) | none | - | - | -| [`GzipCompression`](/docs/modules/compression/api-reference/gzip-compression) | gzip(`.gz`) | size | Native async decode; [small fallback](https://bundlephobia.com/package/pako) | -| [`DeflateCompression`](/docs/modules/compression/api-reference/deflate-compression) | DEFLATE(PKZIP) | size | Native async decode; [small fallback](https://bundlephobia.com/package/pako) | +| [`GzipCompression`](/docs/modules/compression/api-reference/gzip-compression) | gzip(`.gz`) | size | [Small](https://bundlephobia.com/package/pako) | +| [`DeflateCompression`](/docs/modules/compression/api-reference/deflate-compression) | DEFLATE(PKZIP) | size | [Small](https://bundlephobia.com/package/pako) | | [`LZ4Compression`](/docs/modules/compression/api-reference/lz4-compression) | LZ4 | speed ("real-time") | [Medium](https://bundlephobia.com/package/lz4) | -| [`ZstdCompression`](/docs/modules/compression/api-reference/zstd-compression) | Zstandard | speed ("real-time") | Native async decode when available; [large fallback](https://bundlephobia.com/package/zstd-codec) | +| [`ZstdCompression`](/docs/modules/compression/api-reference/zstd-compression) | Zstandard | speed ("real-time") | [Large](https://bundlephobia.com/package/zstd-codec) | | [`SnappyCompression`](/docs/modules/compression/api-reference/snappy-compression) | Snappy(Zippy) | speed ("real-time") | [Small](https://bundlephobia.com/package/snappys) | -| [`BrotliCompression`](/docs/modules/compression/api-reference/brotli-compression) | Brotli | Size, fast decompress, slow compress | Native async decode when available; [large fallback](https://bundlephobia.com/package/brotli) | +| [`BrotliCompression`](/docs/modules/compression/api-reference/brotli-compression) | Brotli | Size, fast decompress, slow compress | [Large](https://bundlephobia.com/package/brotli) | | [`LZOCompression`](/docs/modules/compression/api-reference/lzo-compression) | Lempel-Ziv-Oberheimer | size | Node.js only | ## Compression Formats diff --git a/docs/modules/compression/api-reference/brotli-compression.md b/docs/modules/compression/api-reference/brotli-compression.md index d709df668a..2db22b0234 100644 --- a/docs/modules/compression/api-reference/brotli-compression.md +++ b/docs/modules/compression/api-reference/brotli-compression.md @@ -6,12 +6,6 @@ Compresses / decompresses Brotli encoded data. -Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's native -`DecompressionStream('brotli')` implementation when available, falling back to the existing codec -otherwise. A provided `brotli` module takes precedence over the native stream path. Explicit codec -options, compression, and synchronous decompression keep their existing codec requirements. -From-v5.0 - ## Interface Implements the [`Compression](./compression) API. @@ -20,4 +14,4 @@ Implements the [`Compression](./compression) API. ### `constructor(options?: object)` -`options` is optional for native asynchronous decompression. +`options` is optional when using the built-in Brotli decoder. diff --git a/docs/modules/compression/api-reference/deflate-compression.md b/docs/modules/compression/api-reference/deflate-compression.md index f2fadcc495..30282e4382 100644 --- a/docs/modules/compression/api-reference/deflate-compression.md +++ b/docs/modules/compression/api-reference/deflate-compression.md @@ -6,12 +6,6 @@ Compresses / decompresses DEFLATE encoded data. -Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's native -`DecompressionStream('deflate')` implementation when available. With `raw: true`, loaders.gl -probes `DecompressionStream('deflate-raw')`. It falls back to the existing codec when the native -format is unavailable; other explicit codec options, compression, and synchronous decompression keep -their existing codec requirements. From-v5.0 - ## Interface Implements the [`Compression](./compression) API. diff --git a/docs/modules/compression/api-reference/gzip-compression.md b/docs/modules/compression/api-reference/gzip-compression.md index 47dcf92149..106be7e839 100644 --- a/docs/modules/compression/api-reference/gzip-compression.md +++ b/docs/modules/compression/api-reference/gzip-compression.md @@ -6,11 +6,6 @@ Compresses / decompresses GZIP encoded data. -Default asynchronous `decompress()` and `decompressBatches()` calls use the runtime's native -`DecompressionStream('gzip')` implementation when available, falling back to the existing codec -otherwise. Explicit codec options, compression, and synchronous decompression keep their existing -codec requirements. From-v5.0 - ## Interface Implements the [`Compression](./compression) API. 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 32242a7775..a4910604fc 100644 --- a/docs/modules/compression/api-reference/zstd-compression.md +++ b/docs/modules/compression/api-reference/zstd-compression.md @@ -6,14 +6,10 @@ Compresses / decompresses Zstandard encoded data. -When no `zstd-codec` module is registered, asynchronous `decompress()` and -`decompressBatches()` probe `new DecompressionStream('zstd')` so future runtimes can use native -Zstandard support automatically. Native Zstandard support is not yet widely available. -From-v5.0 - -Inject `zstd-codec` through `options.modules` for broad compatibility. When it is provided, it -takes precedence over the native stream path. `compress()`, `compressSync()`, and -`decompressSync()` continue to require `zstd-codec` in every runtime. +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 @@ -23,6 +19,5 @@ Implements the [`Compression](./compression) API. ### `constructor(options?: object)` -`options` is optional for future native asynchronous decompression. Supply -`{modules: {'zstd-codec': ZstdCodec}}` for broad runtime compatibility or when a -synchronous/compression API is needed. +`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 57747da929..46e486ca89 100644 --- a/docs/modules/parquet/api-reference/parquet-loader.md +++ b/docs/modules/parquet/api-reference/parquet-loader.md @@ -119,11 +119,12 @@ field-level GeoArrow metadata. ## Compressions -Some compression codecs are big and need to be imported explicitly by the application and passed -to the `ParquetLoader`. Without an injected codec, Zstandard pages can use -`new DecompressionStream('zstd')` when a future runtime supports it. Inject `zstd-codec` for broad -runtime compatibility; when provided, loaders.gl uses it instead of the native stream path. LZ4 -still requires `lz4js`. From-v5.0 +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 a84ef39b0c..965e617e63 100644 --- a/docs/modules/splats/api-reference/spz-loader.md +++ b/docs/modules/splats/api-reference/spz-loader.md @@ -17,10 +17,12 @@ ## Usage -SPZ version 4 uses ZSTD-compressed attribute streams. Without an injected codec, async SPZ parsing -can use `new DecompressionStream('zstd')` when a future runtime supports it. Inject `zstd-codec` -through loader options for broad runtime compatibility; when provided, loaders.gl uses it instead -of the native stream path. +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. diff --git a/docs/whats-new.mdx b/docs/whats-new.mdx index 0e180ab307..909b6c5e7d 100644 --- a/docs/whats-new.mdx +++ b/docs/whats-new.mdx @@ -75,7 +75,7 @@ Release Date: 2026 **@loaders.gl/compression** -- Async gzip, deflate, raw deflate, and Brotli decompression now probes and uses the runtime's native `DecompressionStream` before falling back to existing codecs. Zstandard is also probed so future runtimes can use native support automatically; until native Zstandard support becomes widely available, most runtimes still need `zstd-codec` for Zstandard-backed Parquet and SPZ parsing. +- 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** 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 20dd76fecd..9db0c368b5 100644 --- a/modules/compression/src/lib/brotli-compression.ts +++ b/modules/compression/src/lib/brotli-compression.ts @@ -48,19 +48,9 @@ export class BrotliCompression extends Compression { readonly isSupported = true; readonly options: BrotliCompressionOptions; - /** Native Brotli format used for default asynchronous decompression. */ - protected readonly decompressionStreamFormat = 'brotli'; - - /** Only use native decompression when no codec-specific options would be ignored. */ - protected readonly useNativeDecompressionStream: boolean; - - /** Registered Brotli module that takes precedence over native decompression. */ - protected readonly decompressionModuleName = 'brotli'; - constructor(options: BrotliCompressionOptions = {}) { super(options); this.options = options; - this.useNativeDecompressionStream = !options.brotli || Object.keys(options.brotli).length === 0; registerJSModules(options?.modules); } @@ -100,11 +90,6 @@ export class BrotliCompression extends Compression { } async decompress(input: ArrayBuffer): Promise { - const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); - if (nativeOutput) { - return nativeOutput; - } - // On Node.js we can use built-in zlib if (!isBrowser && this.options.brotli?.useZlib) { const buffer = await promisify1(zlib.brotliDecompress)(input); diff --git a/modules/compression/src/lib/compression.ts b/modules/compression/src/lib/compression.ts index 977cc7a674..446bcf6093 100644 --- a/modules/compression/src/lib/compression.ts +++ b/modules/compression/src/lib/compression.ts @@ -3,16 +3,7 @@ // Copyright (c) vis.gl contributors // Compression interface -import { - concatenateArrayBuffersAsync, - getJSModuleOrNull, - registerJSModules -} from '@loaders.gl/loader-utils'; -import { - decompressBatchesWithNativeDecompressionStream, - decompressWithNativeDecompressionStream, - type NativeDecompressionFormat -} from './decompression-stream'; +import {concatenateArrayBuffersAsync, registerJSModules} from '@loaders.gl/loader-utils'; /** Compression options */ export type CompressionOptions = { @@ -27,15 +18,6 @@ export abstract class Compression { abstract readonly contentEncodings: string[]; abstract readonly isSupported: boolean; - /** Native format used for default asynchronous decompression, when available. */ - protected readonly decompressionStreamFormat: NativeDecompressionFormat | undefined = undefined; - - /** Whether default asynchronous decompression can use the native stream path. */ - protected readonly useNativeDecompressionStream: boolean = true; - - /** Registered fallback module that takes precedence over native decompression. */ - protected readonly decompressionModuleName: string | undefined = undefined; - constructor(options?: CompressionOptions) { this.compressBatches = this.compressBatches.bind(this); this.decompressBatches = this.decompressBatches.bind(this); @@ -55,10 +37,6 @@ export abstract class Compression { /** Asynchronously decompress data */ async decompress(input: ArrayBuffer, size?: number): Promise { - const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); - if (nativeOutput) { - return nativeOutput; - } await this.preload(); return this.decompressSync(input, size); } @@ -86,17 +64,6 @@ export abstract class Compression { async *decompressBatches( asyncIterator: AsyncIterable | Iterable ): AsyncIterable { - if (this.decompressionStreamFormat && this.shouldUseNativeDecompressionStream()) { - const outputBatches = decompressBatchesWithNativeDecompressionStream( - asyncIterator, - this.decompressionStreamFormat - ); - if (outputBatches) { - yield* outputBatches; - return; - } - } - // TODO - implement incremental compression const input = await this.concatenate(asyncIterator); yield this.decompress(input); @@ -108,34 +75,6 @@ export abstract class Compression { return concatenateArrayBuffersAsync(asyncIterator); } - /** - * Attempts native asynchronous decompression for classes that declare a stream format. - * - * @param input Compressed input data. - * @returns Decompressed data, or null when native decompression should not be used. - */ - protected async tryDecompressWithNativeDecompressionStream( - input: ArrayBuffer - ): Promise { - if (!this.decompressionStreamFormat || !this.shouldUseNativeDecompressionStream()) { - return null; - } - return await decompressWithNativeDecompressionStream(input, this.decompressionStreamFormat); - } - - /** - * Returns whether native asynchronous decompression should be attempted. - * - * Explicitly registered fallback modules take precedence so applications that provide a - * library do not silently switch implementations when a runtime adds native support. - */ - protected shouldUseNativeDecompressionStream(): boolean { - return ( - this.useNativeDecompressionStream && - (!this.decompressionModuleName || !getJSModuleOrNull(this.decompressionModuleName)) - ); - } - protected improveError(error) { if (!error.message.includes(this.name)) { error.message = `${this.name} ${error.message}`; diff --git a/modules/compression/src/lib/decompression-stream.ts b/modules/compression/src/lib/decompression-stream.ts index 9847fe3626..b877496752 100644 --- a/modules/compression/src/lib/decompression-stream.ts +++ b/modules/compression/src/lib/decompression-stream.ts @@ -2,8 +2,6 @@ // SPDX-License-Identifier: MIT // Copyright (c) vis.gl contributors -import {concatenateArrayBuffersAsync, ensureArrayBuffer} from '@loaders.gl/loader-utils'; - /** * Compression formats that runtimes may expose through DecompressionStream. * @@ -28,7 +26,7 @@ export async function decompressWithNativeDecompressionStream( format: NativeDecompressionFormat ): Promise { const outputBatches = decompressBatchesWithNativeDecompressionStream([input], format); - return outputBatches ? await concatenateArrayBuffersAsync(outputBatches) : null; + return outputBatches ? await concatenateNativeDecompressionBatches(outputBatches) : null; } /** @@ -101,7 +99,7 @@ async function* transformBatchesWithNativeDecompressionStream( outputCompleted = true; break; } - yield ensureArrayBuffer(value); + yield copyExactArrayBuffer(value); } await writePromise; } finally { @@ -114,6 +112,42 @@ async function* transformBatchesWithNativeDecompressionStream( } } +/** + * 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. * diff --git a/modules/compression/src/lib/deflate-compression.ts b/modules/compression/src/lib/deflate-compression.ts index 23c4c46600..e6345ee234 100644 --- a/modules/compression/src/lib/deflate-compression.ts +++ b/modules/compression/src/lib/deflate-compression.ts @@ -5,10 +5,6 @@ // DEFLATE import type {CompressionOptions} from './compression'; import {Compression} from './compression'; -import { - decompressBatchesWithNativeDecompressionStream, - type NativeDecompressionFormat -} from './decompression-stream'; import {isBrowser, toArrayBuffer, promisify1} from '@loaders.gl/loader-utils'; import pako from 'pako'; // https://bundlephobia.com/package/pako import zlib from 'zlib'; @@ -30,25 +26,11 @@ export class DeflateCompression extends Compression { readonly options: DeflateCompressionOptions; - /** Native format matching the configured wrapper type. */ - protected readonly decompressionStreamFormat: NativeDecompressionFormat; - - /** Only use native decompression when no codec-specific options would be ignored. */ - protected readonly useNativeDecompressionStream: boolean; - private _chunks: ArrayBuffer[] = []; constructor(options: DeflateCompressionOptions = {}) { super(options); this.options = options; - this.decompressionStreamFormat = options.raw - ? 'deflate-raw' - : options.deflate?.gzip - ? 'gzip' - : 'deflate'; - const deflateOptions = options.deflate; - this.useNativeDecompressionStream = - !deflateOptions || Object.keys(deflateOptions).every(optionName => optionName === 'gzip'); } async compress(input: ArrayBuffer): Promise { @@ -63,11 +45,6 @@ export class DeflateCompression extends Compression { } async decompress(input: ArrayBuffer): Promise { - const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); - if (nativeOutput) { - return nativeOutput; - } - // On Node.js we can use built-in zlib if (!isBrowser && this.options.deflate?.useZlib) { const buffer = this.options.deflate?.gzip @@ -113,17 +90,6 @@ export class DeflateCompression extends Compression { async *decompressBatches( asyncIterator: AsyncIterable | Iterable ): AsyncIterable { - if (this.decompressionStreamFormat && this.useNativeDecompressionStream) { - const outputBatches = decompressBatchesWithNativeDecompressionStream( - asyncIterator, - this.decompressionStreamFormat - ); - if (outputBatches) { - yield* outputBatches; - return; - } - } - const pakoOptions: pako.InflateOptions = this.options?.deflate || {}; const pakoProcessor = new pako.Inflate(pakoOptions); yield* this.transformBatches(pakoProcessor, asyncIterator); diff --git a/modules/compression/src/lib/zstd-compression.ts b/modules/compression/src/lib/zstd-compression.ts index 7baea362ae..399e3cd1d3 100644 --- a/modules/compression/src/lib/zstd-compression.ts +++ b/modules/compression/src/lib/zstd-compression.ts @@ -7,6 +7,7 @@ import type {CompressionOptions} from './compression'; import {Compression} from './compression'; import { registerJSModules, + checkJSModule, getJSModule, getJSModuleOrNull, ensureArrayBuffer @@ -29,12 +30,6 @@ export class ZstdCompression extends Compression { readonly isSupported = true; readonly options: CompressionOptions; - /** Native Zstandard format used for default asynchronous decompression. */ - protected readonly decompressionStreamFormat = 'zstd'; - - /** Registered Zstandard module that takes precedence over native decompression. */ - protected readonly decompressionModuleName = 'zstd-codec'; - /** * zstd-codec is an injectable dependency due to big size * @param options @@ -47,6 +42,7 @@ export class ZstdCompression extends Compression { async preload(modules: Record = {}): Promise { registerJSModules(modules); + checkJSModule('zstd-codec', this.name); const ZstdCodec = getJSModuleOrNull('zstd-codec'); // eslint-disable-next-line @typescript-eslint/no-misused-promises if (!zstdPromise && ZstdCodec) { @@ -74,13 +70,7 @@ export class ZstdCompression extends Compression { } async decompress(input: ArrayBuffer, size?: number): Promise { - const nativeOutput = await this.tryDecompressWithNativeDecompressionStream(input); - if (nativeOutput) { - return nativeOutput; - } - await this.preload(); - getJSModule('zstd-codec', this.name); const simpleZstd = new zstd.Streaming(); const inputArray = new Uint8Array(input); 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 0431f0804d..eca604bfd5 100644 --- a/modules/compression/test/compression.spec.ts +++ b/modules/compression/test/compression.spec.ts @@ -15,6 +15,10 @@ 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'; @@ -169,7 +173,7 @@ test('compression#batched', async t => { t.end(); }); -test('compression#native DecompressionStream formats', async t => { +test('native decompression#real DecompressionStream formats', async t => { for (const format of Object.keys( NATIVE_DECOMPRESSION_FIXTURES ) as NativeDecompressionTestFormat[]) { @@ -178,31 +182,18 @@ test('compression#native DecompressionStream formats', async t => { continue; } - const restoreModule = - format === 'brotli' - ? removeRegisteredModule('brotli') - : format === 'zstd' - ? removeRegisteredModule('zstd-codec') - : null; const nativeFormats: NativeDecompressionTestFormat[] = []; const restoreDecompressionStream = installRecordingDecompressionStream(nativeFormats); try { - const compression = - format === 'gzip' - ? new GZipCompression() - : format === 'deflate' - ? new DeflateCompression() - : format === 'deflate-raw' - ? new DeflateCompression({raw: true}) - : format === 'brotli' - ? new BrotliCompression() - : new ZstdCompression(); const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES[format]).buffer; - const decompressedData = await compression.decompress(compressedData); + const decompressedData = await decompressWithNativeDecompressionStream( + compressedData, + format + ); t.ok( - compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), + decompressedData && compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), `native atomic ${format} decompression works` ); @@ -211,7 +202,11 @@ test('compression#native DecompressionStream formats', async t => { compressedData.slice(0, splitIndex), compressedData.slice(splitIndex, compressedData.byteLength) ]; - const decompressedBatches = compression.decompressBatches(compressedBatches); + 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), @@ -224,69 +219,40 @@ test('compression#native DecompressionStream formats', async t => { ); } finally { restoreDecompressionStream(); - restoreModule?.(); } } t.end(); }); -test('zstd#native DecompressionStream works without zstd-codec', async t => { +test('native decompression#mocked zstd atomic and batched', async t => { const formats: string[] = []; const restoreDecompressionStream = installMockDecompressionStream({ formats, supportedFormats: ['zstd'] }); - const restoreZstdCodec = removeRegisteredModule('zstd-codec'); try { const inputBatches = [new Uint8Array([1, 2, 3]).buffer, new Uint8Array([4, 5, 6]).buffer]; const inputData = concatenateArrayBuffers(...inputBatches); - const compression = new ZstdCompression(); - const decompressedData = await compression.decompress(inputData); - t.ok(compareArrayBuffers(inputData, decompressedData), 'native atomic zstd needs no codec'); - - const decompressedBatches = compression.decompressBatches(inputBatches); - const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); + const decompressedData = await decompressWithNativeDecompressionStream(inputData, 'zstd'); t.ok( - compareArrayBuffers(inputData, decompressedBatchData), - 'native batched zstd needs no codec' + decompressedData && compareArrayBuffers(inputData, decompressedData), + 'native atomic zstd needs no codec' ); - t.deepEqual(formats, ['zstd', 'zstd'], 'zstd maps to the native zstd format'); - } finally { - restoreZstdCodec(); - restoreDecompressionStream(); - } - t.end(); -}); - -test('zstd#provided zstd-codec bypasses native DecompressionStream', async t => { - const formats: string[] = []; - const restoreDecompressionStream = installMockDecompressionStream({ - formats, - supportedFormats: ['zstd'] - }); - - try { - const inputData = new Uint8Array([1, 2, 3, 4, 5, 6]).buffer; - const compression = new ZstdCompression({modules}); - const compressedData = await compression.compress(inputData); - const decompressedData = await compression.decompress(compressedData); - - t.ok(compareArrayBuffers(inputData, decompressedData), 'zstd codec fallback decompresses data'); - const splitIndex = Math.max(1, Math.floor(compressedData.byteLength / 2)); - const decompressedBatches = compression.decompressBatches([ - compressedData.slice(0, splitIndex), - compressedData.slice(splitIndex, compressedData.byteLength) - ]); + const decompressedBatches = decompressBatchesWithNativeDecompressionStream( + inputBatches, + 'zstd' + ); + t.ok(decompressedBatches, 'native batched zstd stream is created'); const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); t.ok( compareArrayBuffers(inputData, decompressedBatchData), - 'zstd codec fallback decompresses batches' + 'native batched zstd needs no codec' ); - t.deepEqual(formats, [], 'provided zstd codec bypasses the native stream'); + t.deepEqual(formats, ['zstd', 'zstd'], 'zstd maps to the native zstd format'); } finally { restoreDecompressionStream(); } @@ -294,33 +260,26 @@ test('zstd#provided zstd-codec bypasses native DecompressionStream', async t => t.end(); }); -test('brotli#provided module bypasses native DecompressionStream', async t => { +test('native decompression#unsupported formats return null', async t => { const formats: string[] = []; const restoreDecompressionStream = installMockDecompressionStream({ formats, - supportedFormats: ['brotli'] + supportedFormats: [] }); try { - const compression = new BrotliCompression({modules}); - const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES.brotli).buffer; - const decompressedData = await compression.decompress(compressedData); - - t.ok( - compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), - 'provided brotli module decompresses data' + const inputData = new Uint8Array([1, 2, 3]).buffer; + t.equal( + await decompressWithNativeDecompressionStream(inputData, 'zstd'), + null, + 'atomic unsupported format returns null' ); - const splitIndex = Math.max(1, Math.floor(compressedData.byteLength / 2)); - const decompressedBatches = compression.decompressBatches([ - compressedData.slice(0, splitIndex), - compressedData.slice(splitIndex, compressedData.byteLength) - ]); - const decompressedBatchData = await concatenateArrayBuffersAsync(decompressedBatches); - t.ok( - compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedBatchData), - 'provided brotli module decompresses batches' + t.equal( + decompressBatchesWithNativeDecompressionStream([inputData], 'zstd'), + null, + 'batched unsupported format returns null' ); - t.deepEqual(formats, [], 'provided brotli module bypasses the native stream'); + t.deepEqual(formats, ['zstd', 'zstd'], 'both paths probe the requested format'); } finally { restoreDecompressionStream(); } @@ -328,68 +287,27 @@ test('brotli#provided module bypasses native DecompressionStream', async t => { t.end(); }); -test('zstd#native stream failures do not fall back', async t => { +test('native decompression#stream failures propagate', async t => { const restoreDecompressionStream = installMockDecompressionStream({ formats: [], supportedFormats: ['zstd'], failWith: new Error('mock native decompression failed') }); - const restoreZstdCodec = removeRegisteredModule('zstd-codec'); try { const inputData = new Uint8Array([1, 2, 3]).buffer; - const compression = new ZstdCompression(); await t.rejects( - compression.decompress(inputData), + decompressWithNativeDecompressionStream(inputData, 'zstd'), /mock native decompression failed/, 'native stream errors propagate' ); } finally { - restoreZstdCodec(); restoreDecompressionStream(); } t.end(); }); -test('deflate#native format mapping and explicit option fallback', async t => { - const formats: string[] = []; - const restoreDecompressionStream = installMockDecompressionStream({ - formats, - supportedFormats: ['deflate-raw'] - }); - - try { - const inputData = new Uint8Array([1, 2, 3, 4, 5, 6]).buffer; - const rawCompression = new DeflateCompression({raw: true}); - const decompressedRawData = await rawCompression.decompress(inputData); - - t.ok(compareArrayBuffers(inputData, decompressedRawData), 'raw deflate uses native stream'); - t.deepEqual(formats, ['deflate-raw'], 'raw deflate maps to deflate-raw'); - - formats.length = 0; - const compressedData = new DeflateCompression().compressSync(inputData); - const configuredCompression = new DeflateCompression({deflate: {useZlib: true}}); - const decompressedConfiguredData = await configuredCompression.decompress(compressedData); - - t.ok( - compareArrayBuffers(inputData, decompressedConfiguredData), - 'configured deflate uses the existing implementation' - ); - t.deepEqual(formats, [], 'codec-specific options bypass the native stream'); - } finally { - restoreDecompressionStream(); - } - - t.end(); -}); - -test('compression#native constructors accept omitted options', t => { - t.ok(new BrotliCompression(), 'BrotliCompression options are optional'); - t.ok(new ZstdCompression(), 'ZstdCompression options are optional'); - t.end(); -}); - // WORKER TESTS test('gzip#worker', async t => { const {binaryData} = getData(); @@ -543,31 +461,6 @@ function installMockDecompressionStream(options: MockDecompressionStreamOptions) }; } -/** - * 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. * diff --git a/modules/compression/test/decompression-stream.node.spec.ts b/modules/compression/test/decompression-stream.node.spec.ts index 67eb782f00..076980ae30 100644 --- a/modules/compression/test/decompression-stream.node.spec.ts +++ b/modules/compression/test/decompression-stream.node.spec.ts @@ -4,11 +4,9 @@ import test from 'tape-promise/tape'; import { - BrotliCompression, - DeflateCompression, - GZipCompression, - ZstdCompression -} from '@loaders.gl/compression'; + decompressBatchesWithNativeDecompressionStream, + decompressWithNativeDecompressionStream +} from '@loaders.gl/compression/native-decompression'; import {concatenateArrayBuffersAsync} from '@loaders.gl/loader-utils'; import {compareArrayBuffers} from './utils/test-utils'; import { @@ -23,7 +21,7 @@ type MutableGlobalThis = typeof globalThis & { Buffer?: typeof Buffer; }; -test('compression#native DecompressionStream formats in Node.js', async t => { +test('native decompression#real DecompressionStream formats in Node.js', async t => { for (const format of Object.keys( NATIVE_DECOMPRESSION_FIXTURES ) as NativeDecompressionTestFormat[]) { @@ -36,21 +34,14 @@ test('compression#native DecompressionStream formats in Node.js', async t => { const restoreDecompressionStream = installRecordingDecompressionStream(nativeFormats); try { - const compression = - format === 'gzip' - ? new GZipCompression() - : format === 'deflate' - ? new DeflateCompression() - : format === 'deflate-raw' - ? new DeflateCompression({raw: true}) - : format === 'brotli' - ? new BrotliCompression() - : new ZstdCompression(); const compressedData = new Uint8Array(NATIVE_DECOMPRESSION_FIXTURES[format]).buffer; - const decompressedData = await compression.decompress(compressedData); + const decompressedData = await decompressWithNativeDecompressionStream( + compressedData, + format + ); t.ok( - compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), + decompressedData && compareArrayBuffers(NATIVE_DECOMPRESSION_TEST_DATA, decompressedData), `native atomic ${format} decompression works in Node.js` ); @@ -59,7 +50,11 @@ test('compression#native DecompressionStream formats in Node.js', async t => { compressedData.slice(0, splitIndex), compressedData.slice(splitIndex, compressedData.byteLength) ]; - const decompressedBatches = compression.decompressBatches(compressedBatches); + 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), @@ -78,17 +73,15 @@ test('compression#native DecompressionStream formats in Node.js', async t => { t.end(); }); -test('gzip#native DecompressionStream falls back without global Buffer in Node.js', async t => { - const inputData = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]).buffer; - const compression = new GZipCompression(); - const compressedData = compression.compressSync(inputData); +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 compression.decompress(compressedData); - t.ok(compareArrayBuffers(inputData, decompressedData), 'gzip falls back without Buffer'); + const decompressedData = await decompressWithNativeDecompressionStream(compressedData, 'gzip'); + t.equal(decompressedData, null, 'native helper lets callers choose a fallback without Buffer'); } finally { mutableGlobalThis.Buffer = originalBuffer; } 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/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/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 From 08a4127d62fe5854ebcca2508d4f3669f0bd29ad Mon Sep 17 00:00:00 2001 From: Ib Green Date: Tue, 4 Aug 2026 13:18:00 -0400 Subject: [PATCH 6/6] benchmark parquet implementations against hyparquet --- modules/parquet/test/parquet.bench.ts | 199 +++++++++++++++++++++++--- scripts/test.mjs | 2 +- 2 files changed, 177 insertions(+), 24 deletions(-) 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/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]