From f280c4198d4265ea653d0317d8f63e72471155c4 Mon Sep 17 00:00:00 2001 From: ump45nose <52391318+ump45nose@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:06:02 +0800 Subject: [PATCH] docs: document how to register IPC compression codecs --- src/ipc/compression/registry.ts | 58 +++++++++++++++++++++++++++++++++ src/ipc/serialization.ts | 7 ++++ src/ipc/writer.ts | 5 +++ 3 files changed, 70 insertions(+) diff --git a/src/ipc/compression/registry.ts b/src/ipc/compression/registry.ts index af7d819c..a17ba00f 100644 --- a/src/ipc/compression/registry.ts +++ b/src/ipc/compression/registry.ts @@ -18,8 +18,23 @@ import { CompressionType } from '../../fb/compression-type.js'; import { compressionValidators } from './validators.js'; +/** + * A compression codec for IPC record batch body buffers. + * + * Arrow JS does not bundle any compression library and does not register any + * codec by default. To read or write compressed IPC data, register a codec for + * each `CompressionType` you need with {@link compressionRegistry}. + * + * A codec only needs `decode` to read compressed data and only needs `encode` + * to write it. Both work on a single, complete buffer: `encode` must return the + * compressed bytes in the format named by the compression type (an LZ4 *frame* + * for `LZ4_FRAME`, a Zstandard frame for `ZSTD`), and `decode` must return the + * original uncompressed bytes. Both must be synchronous. + */ export interface Codec { + /** Compress one body buffer. Needed to write compressed IPC data. */ encode?(data: Uint8Array): Uint8Array; + /** Decompress one body buffer. Needed to read compressed IPC data. */ decode?(data: Uint8Array): Uint8Array; } @@ -30,6 +45,17 @@ class _CompressionRegistry { this.registry = {}; } + /** + * Register (or replace) the codec used for a compression type. + * + * If the codec has an `encode` method, it is called once on a small test + * buffer and the output is checked for the format's magic number, so an + * error is thrown right away when, for example, a raw LZ4 *block* encoder + * is registered for `LZ4_FRAME`. + * + * @param compression The compression type the codec handles. + * @param codec The codec to use for that compression type. + */ set(compression: CompressionType, codec: Codec) { if (codec?.encode && typeof codec.encode === 'function' && !compressionValidators[compression].isValidCodecEncode(codec)) { throw new Error(`Encoder for ${CompressionType[compression]} is not valid.`); @@ -37,10 +63,42 @@ class _CompressionRegistry { this.registry[compression] = codec; } + /** + * Get the codec registered for a compression type, or `null` if none is. + */ get(compression: CompressionType): Codec | null { return this.registry?.[compression] || null; } } +/** + * The global registry of compression codecs used by the IPC readers and writers. + * + * The registry starts out empty. Readers look up a codec only when they meet a + * compressed record batch or dictionary batch, and throw + * `Record batch is compressed but codec not found` if no codec with a `decode` + * method is registered for its compression type. Writers look up a codec only + * when a `compressionType` is passed (see the writer `compressionType` option + * and {@link tableToIPC}), and throw if no codec with an `encode` method is + * registered for it. Uncompressed data never needs a codec, so you only pay + * for (and load) the compression libraries you register. + * + * Codecs are registered process-wide, so register them once at startup, + * before reading or writing compressed data. + * + * @example + * ```ts + * import { compressionRegistry, CompressionType, tableFromIPC, tableToIPC } from 'apache-arrow'; + * import * as lz4 from 'lz4js'; // any LZ4 *frame* implementation works + * + * compressionRegistry.set(CompressionType.LZ4_FRAME, { + * encode: (data) => lz4.compress(data), + * decode: (data) => lz4.decompress(data), + * }); + * + * const bytes = tableToIPC(table, 'stream', CompressionType.LZ4_FRAME); + * const roundTripped = tableFromIPC(bytes); + * ``` + */ export const compressionRegistry = new _CompressionRegistry(); diff --git a/src/ipc/serialization.ts b/src/ipc/serialization.ts index c437ffeb..5aaa40f5 100644 --- a/src/ipc/serialization.ts +++ b/src/ipc/serialization.ts @@ -33,6 +33,11 @@ type AsyncRecordBatchReaders = AsyncRecordBatchFileRead /** * Deserialize the IPC format into a {@link Table}. This function is a * convenience wrapper for {@link RecordBatchReader}. Opposite of {@link tableToIPC}. + * + * Compressed IPC data (LZ4 frame or ZSTD body buffers) is decompressed + * automatically, but only if a codec for that compression type has been + * registered with {@link compressionRegistry} first; no codec is registered by + * default, and reading a compressed batch without one throws. */ export function tableFromIPC(source: FromArg0 | FromArg2): Table; export function tableFromIPC(source: FromArg1): Promise>; @@ -58,6 +63,8 @@ export function tableFromIPC(input: any): Table | Pr * * @param table The Table to serialize. * @param type Whether to serialize the Table as a file or a stream. + * @param compressionType Optional compression for the record batch body buffers. + * A codec for it must be registered with {@link compressionRegistry} first. */ export function tableToIPC(table: Table, type: 'file' | 'stream' = 'stream', compressionType: CompressionType | null = null): Uint8Array { const writerOptions: RecordBatchStreamWriterOptions = { compressionType }; diff --git a/src/ipc/writer.ts b/src/ipc/writer.ts index 6554498c..dcf2f397 100644 --- a/src/ipc/writer.ts +++ b/src/ipc/writer.ts @@ -55,6 +55,11 @@ export interface RecordBatchStreamWriterOptions { writeLegacyIpcFormat?: boolean; /** * Specifies the optional compression algorithm to use for record batch body buffers. + * + * A codec with an `encode` method must be registered for this compression + * type with {@link compressionRegistry} before writing, otherwise writing a + * batch throws. Buffers that would not get smaller are written uncompressed, + * as the IPC format allows. Not supported with `writeLegacyIpcFormat`. */ compressionType?: CompressionType | null; }