add capability and option flag to treat int96 as a timestamp (#162)

# Problem

There's currently no way to handle INT96 timestamps.

Closes #158

Solution
========
add a flag to enable treating all INT96 fields in a file as timestamps.

Author: j4ys0n

lib/bufferReader.ts

@@ -9,6 +9,7 @@ export interface BufferReaderOptions {
   default_dictionary_size?: number;
   metadata?: FileMetaDataExt;
   rawStatistics?: Statistics;
+  treatInt96AsTimestamp?: boolean;
 }
 
 interface BufferReaderQueueRow {

lib/codec/plain.ts

@@ -146,17 +146,27 @@ function encodeValues_INT96(values: number[]) {
   return buf;
 }
 
-function decodeValues_INT96(cursor: Cursor, count: number) {
+function decodeValues_INT96(cursor: Cursor, count: number, opts?: Options) {
   const values = [];
+  // Default to false for backward compatibility
+  const treatAsTimestamp = opts?.treatInt96AsTimestamp === true;
 
   for (let i = 0; i < count; ++i) {
+    // when treatAsTimestamp is true, low is nanoseconds since midnight
     const low = INT53.readInt64LE(cursor.buffer, cursor.offset);
+    // when treatAsTimestamp is true, high is Julian day
     const high = cursor.buffer.readUInt32LE(cursor.offset + 8);
 
-    if (high === 0xffffffff) {
-      values.push(~-low + 1); // truncate to 64 actual precision
+    if (treatAsTimestamp) {
+      // Convert Julian day and nanoseconds to a timestamp
+      values.push(convertInt96ToTimestamp(high, low));
     } else {
-      values.push(low); // truncate to 64 actual precision
+      // For non-timestamp INT96 values, maintain existing behavior
+      if (high === 0xffffffff) {
+        values.push(~-low + 1); // negative value
+      } else {
+        values.push(low); // positive value
+      }
     }
 
     cursor.offset += 12;
@@ -165,6 +175,30 @@ function decodeValues_INT96(cursor: Cursor, count: number) {
   return values;
 }
 
+/**
+ * Convert INT96 to timestamp
+ * In the Parquet format, INT96 timestamps are stored as:
+ * - The first 8 bytes (low) represent nanoseconds within the day
+ * - The last 4 bytes (high) represent the Julian day
+ *
+ * @param julianDay Julian day number
+ * @param nanosSinceMidnight Nanoseconds since midnight
+ * @returns JavaScript Date object (UTC)
+ */
+function convertInt96ToTimestamp(julianDay: number, nanosSinceMidnight: number | bigint): Date {
+  // Julian day 2440588 corresponds to 1970-01-01 (Unix epoch)
+  const daysSinceEpoch = julianDay - 2440588;
+
+  // Convert days to milliseconds (86,400,000 ms per day)
+  const millisSinceEpoch = daysSinceEpoch * 86400000;
+
+  // Convert nanoseconds to milliseconds
+  const nanosInMillis = Number(BigInt(nanosSinceMidnight) / 1000000n);
+
+  // Create a UTC Date
+  return new Date(millisSinceEpoch + nanosInMillis);
+}
+
 function encodeValues_FLOAT(values: number[]) {
   const buf = Buffer.alloc(4 * values.length);
   for (let i = 0; i < values.length; i++) {
@@ -322,7 +356,7 @@ export const decodeValues = function (type: ValidValueTypes | string, cursor: Cu
       return decodeValues_INT64(cursor, count, opts);
 
     case 'INT96':
-      return decodeValues_INT96(cursor, count);
+      return decodeValues_INT96(cursor, count, opts);
 
     case 'FLOAT':
       return decodeValues_FLOAT(cursor, count);

lib/codec/types.ts

@@ -22,6 +22,7 @@ export interface Options {
   name?: string;
   precision?: number;
   scale?: number;
+  treatInt96AsTimestamp?: boolean;
 }
 
 export interface Cursor {

lib/declare.ts

@@ -212,6 +212,17 @@ export class NewPageHeader extends parquet_thrift.PageHeader {
   headerSize?: number;
 }
 
+export interface BufferReaderOptions {
+  default_dictionary_size?: number;
+  maxLength?: number;
+  maxSpan?: number;
+  queueWait?: number;
+  metadata?: FileMetaDataExt;
+  cache?: boolean;
+  rawStatistics?: boolean;
+  treatInt96AsTimestamp?: boolean; // Default to false for backward compatibility
+}
+
 export interface WriterOptions {
   pageIndex?: boolean;
   pageSize?: number;

lib/reader.ts

@@ -124,6 +124,7 @@ export class ParquetReader {
   envelopeReader: ParquetEnvelopeReader | null;
   metadata: FileMetaDataExt | null;
   schema: parquet_schema.ParquetSchema;
+  treatInt96AsTimestamp: boolean;
 
   /**
    * Open the parquet file pointed to by the specified path and return a new
@@ -199,6 +200,9 @@ export class ParquetReader {
       throw 'invalid parquet version';
     }
 
+    // Default to false for backward compatibility
+    this.treatInt96AsTimestamp = opts.treatInt96AsTimestamp === true;
+
     // If metadata is a json file then we need to convert INT64 and CTIME
     if (metadata.json) {
       const convert = (o: Record<string, any>) => {
@@ -411,6 +415,7 @@ export class ParquetEnvelopeReader {
   default_dictionary_size: number;
   metadata?: FileMetaDataExt;
   schema?: parquet_schema.ParquetSchema;
+  treatInt96AsTimestamp?: boolean;
 
   static async openFile(filePath: string | Buffer | URL, options?: BufferReaderOptions) {
     const fileStat = await parquet_util.fstat(filePath);
@@ -564,6 +569,7 @@ export class ParquetEnvelopeReader {
     this.fileSize = fileSize;
     this.default_dictionary_size = options.default_dictionary_size || 10000000;
     this.metadata = metadata;
+    this.treatInt96AsTimestamp = options.treatInt96AsTimestamp === true;
     if (options.maxLength || options.maxSpan || options.queueWait) {
       const bufferReader = new BufferReader(this, options);
       this.read = (offset, length) => bufferReader.read(offset, length);
@@ -755,6 +761,7 @@ export class ParquetEnvelopeReader {
       compression: compression,
       column: field,
       num_values: metadata.num_values,
+      treatInt96AsTimestamp: this.treatInt96AsTimestamp,
     });
 
     // If this exists and is greater than zero then we need to have an offset
@@ -1046,6 +1053,7 @@ async function decodeDataPage(cursor: Cursor, header: parquet_thrift.PageHeader,
     precision: opts.column!.precision,
     scale: opts.column!.scale,
     name: opts.column!.name,
+    treatInt96AsTimestamp: opts.treatInt96AsTimestamp,
   });
 
   cursor.offset = cursorEnd;
@@ -1109,6 +1117,7 @@ async function decodeDataPageV2(cursor: Cursor, header: parquet_thrift.PageHeade
 
   const values = decodeValues(opts.type!, valueEncoding as ParquetCodec, valuesBufCursor, valueCountNonNull, {
     bitWidth: opts.column!.typeLength!,
+    treatInt96AsTimestamp: opts.treatInt96AsTimestamp,
     ...opts.column!,
   });
 

test/int96_timestamp.test.js

@@ -0,0 +1,70 @@
+'use strict';
+const chai = require('chai');
+const assert = chai.assert;
+const { ParquetReader } = require('../lib/reader');
+
+describe('INT96 timestamp handling', function () {
+  this.timeout(30000); // Increase timeout for URL fetching
+
+  const testUrl =
+    'https://aws-public-blockchain.s3.us-east-2.amazonaws.com/v1.0/eth/traces/date%3D2016-05-22/part-00000-54f4b70c-db10-479c-a117-e3cc760a7e26-c000.snappy.parquet';
+
+  it('should handle INT96 values as numbers by default', async function () {
+    const reader = await ParquetReader.openUrl(testUrl);
+    const cursor = reader.getCursor();
+
+    // Read the first row
+    const row = await cursor.next();
+
+    // Find INT96 columns (if any)
+    const schema = reader.getSchema();
+    const fields = schema.fields;
+
+    // Check if there are any INT96 columns and verify they're numbers
+    let foundInt96 = false;
+    for (const fieldName in fields) {
+      const field = fields[fieldName];
+      if (field.primitiveType === 'INT96') {
+        foundInt96 = true;
+        assert.isNumber(row[fieldName], `Expected ${fieldName} to be a number`);
+      }
+    }
+
+    // If no INT96 columns were found, log a message
+    if (!foundInt96) {
+      console.log('No INT96 columns found in the test file');
+    }
+
+    await reader.close();
+  });
+
+  it('should convert INT96 values to timestamps when option is enabled', async function () {
+    const reader = await ParquetReader.openUrl(testUrl, { treatInt96AsTimestamp: true });
+    const cursor = reader.getCursor();
+
+    // Read the first row
+    const row = await cursor.next();
+
+    // Find INT96 columns (if any)
+    const schema = reader.getSchema();
+    const fields = schema.fields;
+
+    // Check if there are any INT96 columns and verify they're Date objects
+    let foundInt96 = false;
+    for (const fieldName in fields) {
+      const field = fields[fieldName];
+      if (field.primitiveType === 'INT96') {
+        foundInt96 = true;
+        assert.instanceOf(row[fieldName], Date, `Expected ${fieldName} to be a Date object`);
+        console.log(`${fieldName} timestamp:`, row[fieldName]);
+      }
+    }
+
+    // If no INT96 columns were found, log a message
+    if (!foundInt96) {
+      console.log('No INT96 columns found in the test file');
+    }
+
+    await reader.close();
+  });
+});