diff --git a/lib/codec/delta_binary_packed.ts b/lib/codec/delta_binary_packed.ts new file mode 100644 index 00000000..afd0d657 --- /dev/null +++ b/lib/codec/delta_binary_packed.ts @@ -0,0 +1,333 @@ +import { Cursor, Options } from './types'; + +const DEFAULT_BLOCK_SIZE = 128; +const DEFAULT_MINI_BLOCK_COUNT = 4; +const INT32_MIN = -2147483648n; +const INT32_MAX = 2147483647n; +const INT64_MIN = -(1n << 63n); +const INT64_MAX = (1n << 63n) - 1n; + +type IntegerValue = number | bigint; + +function assertIntegerType(type: string) { + if (type !== 'INT32' && type !== 'INT64') { + throw new Error('unsupported type: ' + type); + } +} + +function normalizeIntegerValue(type: string, value: IntegerValue) { + if (typeof value === 'number' && (!Number.isFinite(value) || !Number.isInteger(value))) { + throw new Error(`${type} value must be a finite integer`); + } + + const normalized = typeof value === 'bigint' ? value : BigInt(value); + + if (type === 'INT32' && (normalized < INT32_MIN || normalized > INT32_MAX)) { + throw new Error('INT32 value out of range'); + } + + if (type === 'INT64' && (normalized < INT64_MIN || normalized > INT64_MAX)) { + throw new Error('INT64 value out of range'); + } + + return normalized; +} + +function bitWidthForType(type: string) { + return type === 'INT32' ? 32n : 64n; +} + +function toSignedInteger(value: bigint, bits: bigint) { + const range = 1n << bits; + const midpoint = 1n << (bits - 1n); + const unsignedValue = ((value % range) + range) % range; + + return unsignedValue >= midpoint ? unsignedValue - range : unsignedValue; +} + +function toOutputValue(type: string, value: bigint) { + const normalized = toSignedInteger(value, bitWidthForType(type)); + + if (type === 'INT64') { + return normalized; + } + + return Number(normalized); +} + +function readUnsignedVarint(cursor: Cursor) { + let result = 0n; + let shift = 0n; + + while (cursor.offset < cursor.buffer.length) { + const byte = BigInt(cursor.buffer[cursor.offset]); + cursor.offset += 1; + result |= (byte & 0x7fn) << shift; + + if ((byte & 0x80n) === 0n) { + return result; + } + + shift += 7n; + if (shift > 70n) { + throw new Error('invalid DELTA_BINARY_PACKED encoding'); + } + } + + throw new Error('invalid DELTA_BINARY_PACKED encoding'); +} + +function writeUnsignedVarint(value: bigint) { + if (value < 0n) { + throw new Error('varint value must be unsigned'); + } + + const bytes = []; + let remaining = value; + + while (remaining >= 0x80n) { + bytes.push(Number((remaining & 0x7fn) | 0x80n)); + remaining >>= 7n; + } + + bytes.push(Number(remaining)); + return Buffer.from(bytes); +} + +function decodeZigZag(value: bigint) { + return (value & 1n) === 0n ? value >> 1n : -((value + 1n) >> 1n); +} + +function encodeZigZag(value: bigint) { + return value >= 0n ? value << 1n : (-value << 1n) - 1n; +} + +function readZigZagVarint(cursor: Cursor) { + return decodeZigZag(readUnsignedVarint(cursor)); +} + +function writeZigZagVarint(value: bigint) { + return writeUnsignedVarint(encodeZigZag(value)); +} + +function toSafeCount(value: bigint, name: string) { + if (value > BigInt(Number.MAX_SAFE_INTEGER)) { + throw new Error(`${name} is too large`); + } + + return Number(value); +} + +function validateBlockLayout(blockSize: number, miniBlockCount: number) { + if ( + !Number.isInteger(blockSize) || + !Number.isInteger(miniBlockCount) || + blockSize <= 0 || + miniBlockCount <= 0 || + blockSize % miniBlockCount !== 0 + ) { + throw new Error('invalid DELTA_BINARY_PACKED block layout'); + } +} + +function bitLength(value: bigint) { + if (value < 0n) { + throw new Error('bit-packed values must be unsigned'); + } + + let bits = 0; + let remaining = value; + while (remaining > 0n) { + bits += 1; + remaining >>= 1n; + } + + return bits; +} + +function maxBitWidth(values: bigint[]) { + let width = 0; + for (const value of values) { + width = Math.max(width, bitLength(value)); + } + + return width; +} + +function bitMask(bitWidth: number) { + return (1n << BigInt(bitWidth)) - 1n; +} + +function packMiniBlock(values: bigint[], bitWidth: number) { + const buf = Buffer.alloc(Math.ceil((values.length * bitWidth) / 8)); + + if (bitWidth === 0) { + return buf; + } + + let accumulator = 0n; + let bitsInAccumulator = 0; + let byteOffset = 0; + + for (const value of values) { + accumulator = (accumulator << BigInt(bitWidth)) | value; + bitsInAccumulator += bitWidth; + + while (bitsInAccumulator >= 8) { + bitsInAccumulator -= 8; + buf[byteOffset] = Number((accumulator >> BigInt(bitsInAccumulator)) & 0xffn); + byteOffset += 1; + accumulator &= bitMask(bitsInAccumulator); + } + } + + if (bitsInAccumulator > 0) { + buf[byteOffset] = Number((accumulator << BigInt(8 - bitsInAccumulator)) & 0xffn); + } + + return buf; +} + +function unpackMiniBlock(cursor: Cursor, count: number, bitWidth: number) { + const byteLength = Math.ceil((count * bitWidth) / 8); + if (cursor.offset + byteLength > cursor.buffer.length) { + throw new Error('invalid DELTA_BINARY_PACKED encoding'); + } + + if (bitWidth === 0) { + return new Array(count).fill(0n); + } + + const values = []; + let accumulator = 0n; + let bitsInAccumulator = 0; + let byteOffset = cursor.offset; + + for (let valueIndex = 0; valueIndex < count; valueIndex++) { + while (bitsInAccumulator < bitWidth) { + accumulator = (accumulator << 8n) | BigInt(cursor.buffer[byteOffset]); + bitsInAccumulator += 8; + byteOffset += 1; + } + + bitsInAccumulator -= bitWidth; + const value = (accumulator >> BigInt(bitsInAccumulator)) & bitMask(bitWidth); + values.push(value); + accumulator &= bitMask(bitsInAccumulator); + } + + cursor.offset += byteLength; + return values; +} + +function getBlockLayout(opts?: Options) { + const blockSize = opts?.deltaBinaryPackedBlockSize ?? opts?.blockSize ?? DEFAULT_BLOCK_SIZE; + const miniBlockCount = opts?.deltaBinaryPackedMiniBlockCount ?? opts?.miniBlockCount ?? DEFAULT_MINI_BLOCK_COUNT; + + validateBlockLayout(blockSize, miniBlockCount); + return { blockSize, miniBlockCount }; +} + +export const encodeValues = function (type: string, values: IntegerValue[], opts?: Options) { + assertIntegerType(type); + const { blockSize, miniBlockCount } = getBlockLayout(opts); + + const normalizedValues = values.map((value) => normalizeIntegerValue(type, value)); + const header = [ + writeUnsignedVarint(BigInt(blockSize)), + writeUnsignedVarint(BigInt(miniBlockCount)), + writeUnsignedVarint(BigInt(normalizedValues.length)), + ]; + + if (normalizedValues.length === 0) { + return Buffer.concat(header); + } + + header.push(writeZigZagVarint(normalizedValues[0])); + + const buffers = [...header]; + const valuesPerMiniBlock = blockSize / miniBlockCount; + const typeBitWidth = bitWidthForType(type); + const deltas = []; + for (let i = 1; i < normalizedValues.length; i++) { + deltas.push(toSignedInteger(normalizedValues[i] - normalizedValues[i - 1], typeBitWidth)); + } + + for (let offset = 0; offset < deltas.length; offset += blockSize) { + const blockDeltas = deltas.slice(offset, offset + blockSize); + const minDelta = blockDeltas.reduce((min, value) => (value < min ? value : min), blockDeltas[0]); + const adjustedDeltas = blockDeltas.map((value) => value - minDelta); + const bitWidths = Buffer.alloc(miniBlockCount); + const miniBlocks = []; + + buffers.push(writeZigZagVarint(minDelta)); + + for (let miniBlockIndex = 0; miniBlockIndex < miniBlockCount; miniBlockIndex++) { + const start = miniBlockIndex * valuesPerMiniBlock; + const miniBlockValues = adjustedDeltas.slice(start, start + valuesPerMiniBlock); + while (miniBlockValues.length < valuesPerMiniBlock) { + miniBlockValues.push(0n); + } + + const bitWidth = maxBitWidth(miniBlockValues); + bitWidths[miniBlockIndex] = bitWidth; + miniBlocks.push(packMiniBlock(miniBlockValues, bitWidth)); + } + + buffers.push(bitWidths, ...miniBlocks); + } + + return Buffer.concat(buffers); +}; + +export const decodeValues = function (type: string, cursor: Cursor, count: number, _opts?: Options) { + assertIntegerType(type); + + const blockSize = toSafeCount(readUnsignedVarint(cursor), 'block size'); + const miniBlockCount = toSafeCount(readUnsignedVarint(cursor), 'mini block count'); + const totalValueCount = toSafeCount(readUnsignedVarint(cursor), 'value count'); + + validateBlockLayout(blockSize, miniBlockCount); + + if (totalValueCount !== count) { + throw new Error(`DELTA_BINARY_PACKED value count ${totalValueCount} does not match requested count ${count}`); + } + + if (totalValueCount === 0) { + return []; + } + + const values = []; + const valuesPerMiniBlock = blockSize / miniBlockCount; + const typeBitWidth = bitWidthForType(type); + let previousValue = readZigZagVarint(cursor); + values.push(toOutputValue(type, previousValue)); + + while (values.length < totalValueCount) { + const minDelta = readZigZagVarint(cursor); + if (cursor.offset + miniBlockCount > cursor.buffer.length) { + throw new Error('invalid DELTA_BINARY_PACKED encoding'); + } + + const bitWidths = cursor.buffer.subarray(cursor.offset, cursor.offset + miniBlockCount); + cursor.offset += miniBlockCount; + + for (const bitWidth of bitWidths) { + if (values.length === totalValueCount) { + break; + } + + const adjustedDeltas = unpackMiniBlock(cursor, valuesPerMiniBlock, bitWidth); + for (const adjustedDelta of adjustedDeltas) { + if (values.length === totalValueCount) { + break; + } + + previousValue = toSignedInteger(previousValue + minDelta + adjustedDelta, typeBitWidth); + values.push(toOutputValue(type, previousValue)); + } + } + } + + return values; +}; diff --git a/lib/codec/index.ts b/lib/codec/index.ts index f32072c6..5518cba1 100644 --- a/lib/codec/index.ts +++ b/lib/codec/index.ts @@ -2,3 +2,4 @@ export * as PLAIN from './plain'; export * as RLE from './rle'; export * as PLAIN_DICTIONARY from './plain_dictionary'; export * as RLE_DICTIONARY from './plain_dictionary'; +export * as DELTA_BINARY_PACKED from './delta_binary_packed'; diff --git a/lib/codec/types.ts b/lib/codec/types.ts index d83438b2..3cdc3730 100644 --- a/lib/codec/types.ts +++ b/lib/codec/types.ts @@ -18,6 +18,10 @@ export interface Options { num_values?: number; rLevelMax?: number; dLevelMax?: number; + blockSize?: number; + miniBlockCount?: number; + deltaBinaryPackedBlockSize?: number; + deltaBinaryPackedMiniBlockCount?: number; type?: string; name?: string; precision?: number; diff --git a/lib/declare.ts b/lib/declare.ts index 68794548..a7414ee8 100644 --- a/lib/declare.ts +++ b/lib/declare.ts @@ -15,7 +15,7 @@ import SplitBlockBloomFilter from './bloom/sbbf'; import { createSBBFParams } from './bloomFilterIO/bloomFilterWriter'; import Int64 from 'node-int64'; -export type ParquetCodec = 'PLAIN' | 'RLE'; +export type ParquetCodec = 'PLAIN' | 'RLE' | 'DELTA_BINARY_PACKED'; export type ParquetCompression = 'UNCOMPRESSED' | 'GZIP' | 'SNAPPY' | 'LZO' | 'BROTLI' | 'LZ4'; export type RepetitionType = 'REQUIRED' | 'OPTIONAL' | 'REPEATED'; export type ParquetType = PrimitiveType | OriginalType; diff --git a/test/codec_delta_binary_packed.test.js b/test/codec_delta_binary_packed.test.js new file mode 100644 index 00000000..42b5683e --- /dev/null +++ b/test/codec_delta_binary_packed.test.js @@ -0,0 +1,118 @@ +'use strict'; +const chai = require('chai'); +const assert = chai.assert; +const parquetCodecDeltaBinaryPacked = require('../lib/codec/delta_binary_packed'); + +function roundTrip(type, expected, opts = {}) { + const buf = parquetCodecDeltaBinaryPacked.encodeValues(type, expected, opts); + return parquetCodecDeltaBinaryPacked.decodeValues(type, { buffer: buf, offset: 0 }, expected.length, opts); +} + +describe('ParquetCodec::DELTA_BINARY_PACKED', function () { + it('should decode constant delta values', function () { + const vals = parquetCodecDeltaBinaryPacked.decodeValues( + 'INT32', + { + buffer: Buffer.from([0x80, 0x01, 0x04, 0x05, 0x14, 0x02, 0x00, 0x00, 0x00, 0x00]), + offset: 0, + }, + 5, + {} + ); + + assert.deepEqual(vals, [10, 11, 12, 13, 14]); + }); + + it('should decode bit-packed mini block values', function () { + const vals = parquetCodecDeltaBinaryPacked.decodeValues( + 'INT32', + { + buffer: Buffer.from([ + 0x80, 0x01, 0x04, 0x05, 0x0a, 0x04, 0x02, 0x00, 0x00, 0x00, 0x1b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + ]), + offset: 0, + }, + 5, + {} + ); + + assert.deepEqual(vals, [5, 7, 10, 14, 19]); + }); + + it('should encode and decode INT32 values', function () { + const expected = [1, 2, 4, 7, 11, 16, 22, 29, 37, 46]; + const vals = roundTrip('INT32', expected); + + assert.deepEqual(vals, expected); + }); + + it('should encode and decode INT64 values', function () { + const expected = [1n, 4n, 10n, 19n, 31n]; + const vals = roundTrip('INT64', expected); + + assert.deepEqual(vals, expected); + }); + + it('should encode and decode negative values and deltas', function () { + const expected = [-10, -7, -8, -20, 0, -1]; + const vals = roundTrip('INT32', expected); + + assert.deepEqual(vals, expected); + }); + + it('should encode and decode empty input', function () { + const vals = roundTrip('INT32', []); + + assert.deepEqual(vals, []); + }); + + it('should encode and decode a single value', function () { + const vals = roundTrip('INT32', [-42]); + + assert.deepEqual(vals, [-42]); + }); + + it('should encode and decode custom block layouts', function () { + const expected = [3, 8, 13, 21, 34, 55, 54, 53, 52, 80]; + const opts = { deltaBinaryPackedBlockSize: 8, deltaBinaryPackedMiniBlockCount: 2 }; + const vals = roundTrip('INT32', expected, opts); + + assert.deepEqual(vals, expected); + }); + + it('should reject invalid custom block layouts', function () { + assert.throws( + () => + parquetCodecDeltaBinaryPacked.encodeValues('INT32', [1, 2, 3], { + deltaBinaryPackedBlockSize: 7, + deltaBinaryPackedMiniBlockCount: 2, + }), + /invalid DELTA_BINARY_PACKED block layout/ + ); + + assert.throws( + () => + parquetCodecDeltaBinaryPacked.encodeValues('INT32', [1, 2, 3], { + deltaBinaryPackedBlockSize: 0, + deltaBinaryPackedMiniBlockCount: 2, + }), + /invalid DELTA_BINARY_PACKED block layout/ + ); + }); + + it('should reject non-integer numeric values', function () { + assert.throws( + () => parquetCodecDeltaBinaryPacked.encodeValues('INT32', [1, 1.2], {}), + /INT32 value must be a finite integer/ + ); + }); + + it('should report requested count mismatches', function () { + const buf = parquetCodecDeltaBinaryPacked.encodeValues('INT32', [1, 2, 3], {}); + + assert.throws( + () => parquetCodecDeltaBinaryPacked.decodeValues('INT32', { buffer: buf, offset: 0 }, 2, {}), + /value count 3 does not match requested count 2/ + ); + }); +}); diff --git a/test/reference-test/first-record.ts b/test/reference-test/first-record.ts index acf9eafd..aab2bb21 100644 --- a/test/reference-test/first-record.ts +++ b/test/reference-test/first-record.ts @@ -140,6 +140,91 @@ export function getFirstRecord(filename: string): [true, any] | [false, null] { return [true, { a: 50462976, b: 1734763876 }]; case 'datapage_v1-uncompressed-checksum.parquet': return [true, { a: 50462976, b: 1734763876 }]; + case 'datapage_v2.snappy.parquet': + return [ + true, + { + a: 'abc', + b: 1, + c: 2, + d: false, + e: { + list: [{ element: 1 }, { element: 2 }, { element: 3 }], + }, + }, + ]; + case 'delta_binary_packed.parquet': + return [ + true, + { + bitwidth0: 6374628540732951412n, + bitwidth1: 0n, + bitwidth2: 0n, + bitwidth3: 0n, + bitwidth4: 0n, + bitwidth5: 0n, + bitwidth6: 0n, + bitwidth7: 0n, + bitwidth8: 0n, + bitwidth9: 0n, + bitwidth10: 0n, + bitwidth11: 0n, + bitwidth12: 0n, + bitwidth13: 0n, + bitwidth14: 0n, + bitwidth15: 0n, + bitwidth16: 0n, + bitwidth17: 0n, + bitwidth18: 0n, + bitwidth19: 0n, + bitwidth20: 0n, + bitwidth21: 0n, + bitwidth22: 0n, + bitwidth23: 0n, + bitwidth24: 0n, + bitwidth25: 0n, + bitwidth26: 0n, + bitwidth27: 0n, + bitwidth28: 0n, + bitwidth29: 0n, + bitwidth30: 0n, + bitwidth31: 0n, + bitwidth32: 0n, + bitwidth33: 0n, + bitwidth34: 0n, + bitwidth35: 0n, + bitwidth36: 0n, + bitwidth37: 0n, + bitwidth38: 0n, + bitwidth39: 0n, + bitwidth40: 0n, + bitwidth41: 0n, + bitwidth42: 0n, + bitwidth43: 0n, + bitwidth44: 0n, + bitwidth45: 0n, + bitwidth46: 0n, + bitwidth47: 0n, + bitwidth48: 0n, + bitwidth49: 0n, + bitwidth50: 0n, + bitwidth51: 0n, + bitwidth52: 0n, + bitwidth53: 0n, + bitwidth54: 0n, + bitwidth55: 0n, + bitwidth56: 0n, + bitwidth57: 0n, + bitwidth58: 0n, + bitwidth59: 0n, + bitwidth60: 0n, + bitwidth61: 0n, + bitwidth62: 0n, + bitwidth63: 0n, + bitwidth64: 0n, + int_value: -2070986743, + }, + ]; case 'dict-page-offset-zero.parquet': return [ true, @@ -615,8 +700,6 @@ export function getFirstRecord(filename: string): [true, any] | [false, null] { ]; // Not supported - see read-all-test.ts // case ("byte_stream_split.zstd.parquet"): - // case ("datapage_v2.snappy.parquet"): - // case ("delta_binary_packed.parquet"): // case ("delta_byte_array.parquet"): // case ("delta_encoding_optional_column.parquet"): // case ("delta_encoding_required_column.parquet"): diff --git a/test/reference-test/read-all.test.ts b/test/reference-test/read-all.test.ts index 56e75293..e909e5f9 100644 --- a/test/reference-test/read-all.test.ts +++ b/test/reference-test/read-all.test.ts @@ -21,12 +21,10 @@ export const unsupported = [ 'nested_structs.rust.parquet', // ZSTD unsupported 'non_hadoop_lz4_compressed.parquet', // ZSTD unsupported 'rle_boolean_encoding.parquet', // BUG?: https://github.com/LibertyDSNP/parquetjs/issues/113 - 'datapage_v2.snappy.parquet', // DELTA_BINARY_PACKED unsupported - 'delta_binary_packed.parquet', // DELTA_BINARY_PACKED unsupported 'delta_byte_array.parquet', // DELTA_BYTE_ARRAY unsupported - 'delta_encoding_optional_column.parquet', // DELTA_BINARY_PACKED unsupported - 'delta_encoding_required_column.parquet', // DELTA_BINARY_PACKED unsupported - 'delta_length_byte_array.parquet', // ZSTD unsupported, DELTA_BINARY_PACKED unsupported + 'delta_encoding_optional_column.parquet', // DELTA_BYTE_ARRAY unsupported + 'delta_encoding_required_column.parquet', // DELTA_BYTE_ARRAY unsupported + 'delta_length_byte_array.parquet', // ZSTD unsupported, DELTA_LENGTH_BYTE_ARRAY unsupported 'large_string_map.brotli.parquet', // Fails as the large string is > 1 GB ];