Estimate distinct values in CSV or JSONL columns while streaming in Node.js using a bounded-memory bitmap and linear counting, with nested JSON paths and no dependencies.

node-profile-approx-distinct-counts.js

#!/usr/bin/env node
/**
 * Estimate distinct value counts in CSV or JSONL while streaming.
 *
 * Uses a fixed-size bitmap and linear counting:
 *
 *   estimated cardinality = -m × ln(V / m)
 *
 * where:
 *   m = total number of bitmap bits
 *   V = number of unset bits
 *
 * Features:
 * - Bounded memory per selected column
 * - Streams CSV or JSONL input
 * - Supports nested JSONL paths such as user.country
 * - Tracks missing, empty, and observed values
 * - Uses SHA-256 from Node's built-in crypto module
 * - No third-party dependencies
 *
 * Usage:
 *   node node-profile-approx-distinct-counts.js data.csv \
 *     --format=csv \
 *     --columns=country,status,user_id
 *
 *   cat data.jsonl | node node-profile-approx-distinct-counts.js \
 *     --format=jsonl \
 *     --columns=user.country,status,user.id \
 *     --bitmap-bits=1048576
 *
 * Options:
 *   --format=csv|jsonl   Input format (default: jsonl)
 *   --columns=a,b,c      Columns to profile
 *   --bitmap-bits=N      Bitmap size per column in bits
 *                        (default: 1048576, or 128 KiB per column)
 *   --delimiter=,        CSV delimiter (default: comma)
 *
 * Notes:
 * - The estimate is approximate.
 * - Accuracy decreases as the bitmap becomes saturated.
 * - The report includes bitmap occupancy to help identify saturation.
 * - CSV records with embedded newlines inside quoted fields are not supported.
 */

const crypto = require("crypto");
const fs = require("fs");
const readline = require("readline");

function usage() {
  console.error(`Usage:
  node node-profile-approx-distinct-counts.js <input.csv> \\
    --format=csv \\
    --columns=country,status,user_id

  cat input.jsonl | node node-profile-approx-distinct-counts.js \\
    --format=jsonl \\
    --columns=user.country,status,user.id \\
    --bitmap-bits=1048576

Options:
  --format=csv|jsonl   Input format (default: jsonl)
  --columns=a,b,c      Columns to profile
  --bitmap-bits=N      Bitmap bits per column (default: 1048576)
  --delimiter=,        CSV delimiter (default: ,)
`);
}

function parseArgs(argv) {
  const args = {
    file: null,
    format: "jsonl",
    columns: null,
    bitmapBits: 1_048_576,
    delimiter: ",",
    help: false,
  };

  for (const arg of argv.slice(2)) {
    if (!args.file && !arg.startsWith("--")) {
      args.file = arg;
      continue;
    }

    if (arg.startsWith("--format=")) {
      args.format = arg.slice("--format=".length).toLowerCase();
      continue;
    }

    if (arg.startsWith("--columns=")) {
      args.columns = arg
        .slice("--columns=".length)
        .split(",")
        .map((column) => column.trim())
        .filter(Boolean);
      continue;
    }

    if (arg.startsWith("--bitmap-bits=")) {
      args.bitmapBits = Number(
        arg.slice("--bitmap-bits=".length)
      );
      continue;
    }

    if (arg.startsWith("--delimiter=")) {
      args.delimiter = decodeDelimiter(
        arg.slice("--delimiter=".length)
      );
      continue;
    }

    if (arg === "-h" || arg === "--help") {
      args.help = true;
      continue;
    }

    throw new Error(`Unknown argument: ${arg}`);
  }

  return args;
}

function decodeDelimiter(value) {
  const aliases = {
    "\\t": "\t",
    tab: "\t",
    comma: ",",
    pipe: "|",
    semicolon: ";",
  };

  return aliases[value.toLowerCase()] ?? value;
}

/**
 * Parse one CSV-like line.
 *
 * Supports:
 * - Quoted fields
 * - Delimiters inside quoted fields
 * - Escaped quotes represented as ""
 *
 * Does not support quoted fields containing physical newlines.
 */
function parseDelimitedLine(line, delimiter = ",") {
  const fields = [];
  let current = "";
  let inQuotes = false;

  for (let i = 0; i < line.length; i += 1) {
    const char = line[i];

    if (inQuotes) {
      if (char === '"') {
        if (line[i + 1] === '"') {
          current += '"';
          i += 1;
        } else {
          inQuotes = false;
        }
      } else {
        current += char;
      }

      continue;
    }

    if (char === '"') {
      inQuotes = true;
      continue;
    }

    if (line.startsWith(delimiter, i)) {
      fields.push(current);
      current = "";
      i += delimiter.length - 1;
      continue;
    }

    current += char;
  }

  if (inQuotes) {
    throw new Error("Unclosed quoted field");
  }

  fields.push(current);
  return fields;
}

function getByPath(object, keyPath) {
  let current = object;

  for (const part of keyPath.split(".")) {
    if (
      current === null ||
      typeof current !== "object" ||
      !Object.prototype.hasOwnProperty.call(current, part)
    ) {
      return {
        exists: false,
        value: undefined,
      };
    }

    current = current[part];
  }

  return {
    exists: true,
    value: current,
  };
}

function normalizeValue(value) {
  if (value === null) {
    return "null:null";
  }

  if (Array.isArray(value)) {
    return `array:${stableStringify(value)}`;
  }

  if (typeof value === "object") {
    return `object:${stableStringify(value)}`;
  }

  return `${typeof value}:${String(value)}`;
}

/**
 * Produce stable JSON text by sorting object keys recursively.
 *
 * This ensures objects with the same content but different key order hash
 * to the same value.
 */
function stableStringify(value) {
  if (Array.isArray(value)) {
    return `[${value.map(stableStringify).join(",")}]`;
  }

  if (value && typeof value === "object") {
    const entries = Object.keys(value)
      .sort()
      .map(
        (key) =>
          `${JSON.stringify(key)}:${stableStringify(value[key])}`
      );

    return `{${entries.join(",")}}`;
  }

  return JSON.stringify(value);
}

/**
 * Hash a value and map it to one bitmap position.
 *
 * SHA-256 is built into Node.js and provides stable distribution.
 */
function bitmapIndex(value, bitmapBits) {
  const digest = crypto
    .createHash("sha256")
    .update(value)
    .digest();

  const high = digest.readUInt32BE(0);
  const low = digest.readUInt32BE(4);

  const hash =
    (BigInt(high) << 32n) |
    BigInt(low);

  return Number(hash % BigInt(bitmapBits));
}

function createColumnState(bitmapBits) {
  return {
    bitmapBits,
    bitmap: Buffer.alloc(Math.ceil(bitmapBits / 8)),
    occupiedBits: 0,
    totalRows: 0,
    observedValues: 0,
    missing: 0,
    empty: 0,
  };
}

function createStates(columns, bitmapBits) {
  return Object.fromEntries(
    columns.map((column) => [
      column,
      createColumnState(bitmapBits),
    ])
  );
}

function isEmpty(value) {
  return (
    value === null ||
    value === undefined ||
    (typeof value === "string" && value.trim() === "")
  );
}

function setBitmapBit(state, bitIndex) {
  const byteIndex = Math.floor(bitIndex / 8);
  const mask = 1 << (bitIndex % 8);

  if ((state.bitmap[byteIndex] & mask) === 0) {
    state.bitmap[byteIndex] |= mask;
    state.occupiedBits += 1;
  }
}

function updateState(state, value, exists) {
  state.totalRows += 1;

  if (!exists) {
    state.missing += 1;
    return;
  }

  if (isEmpty(value)) {
    state.empty += 1;
    return;
  }

  state.observedValues += 1;

  const normalized = normalizeValue(value);
  const index = bitmapIndex(
    normalized,
    state.bitmapBits
  );

  setBitmapBit(state, index);
}

/**
 * Estimate cardinality using linear counting.
 *
 * estimate = -m * ln(V / m)
 *
 * When the bitmap is completely saturated, the estimate tends toward
 * infinity. In that case, null is returned and the report marks the column
 * as saturated.
 */
function estimateDistinct(state) {
  const unsetBits =
    state.bitmapBits - state.occupiedBits;

  if (unsetBits <= 0) {
    return null;
  }

  return (
    -state.bitmapBits *
    Math.log(unsetBits / state.bitmapBits)
  );
}

function roundNumber(value, decimalPlaces = 2) {
  if (value === null || !Number.isFinite(value)) {
    return value;
  }

  return Number(value.toFixed(decimalPlaces));
}

function buildReport(states) {
  return Object.entries(states).map(
    ([column, state]) => {
      const estimate = estimateDistinct(state);
      const occupancy =
        state.occupiedBits / state.bitmapBits;

      return {
        column,
        total_rows: state.totalRows,
        observed_values: state.observedValues,
        missing: state.missing,
        empty: state.empty,
        approximate_distinct: roundNumber(
          estimate,
          0
        ),
        bitmap_bits: state.bitmapBits,
        bitmap_bytes: state.bitmap.length,
        occupied_bits: state.occupiedBits,
        bitmap_occupancy_pct: roundNumber(
          occupancy * 100,
          2
        ),
        saturated: estimate === null,
        warning:
          occupancy >= 0.9
            ? "Bitmap is highly saturated; increase --bitmap-bits."
            : occupancy >= 0.7
              ? "Bitmap occupancy is high; estimate accuracy may be reduced."
              : null,
      };
    }
  );
}

async function runCSV(input, args) {
  const reader = readline.createInterface({
    input,
    crlfDelay: Infinity,
  });

  const states = createStates(
    args.columns,
    args.bitmapBits
  );

  let headers = null;
  let columnIndexes = null;
  let lineNumber = 0;

  for await (const line of reader) {
    lineNumber += 1;

    if (!line.trim()) {
      continue;
    }

    let fields;

    try {
      fields = parseDelimitedLine(
        line,
        args.delimiter
      );
    } catch (error) {
      throw new Error(
        `Invalid delimited record on line ${lineNumber}: ${error.message}`
      );
    }

    if (headers === null) {
      headers = fields.map((header) =>
        header.trim()
      );

      const duplicates = headers.filter(
        (header, index) =>
          headers.indexOf(header) !== index
      );

      if (duplicates.length > 0) {
        throw new Error(
          `Duplicate CSV header(s): ${[
            ...new Set(duplicates),
          ].join(", ")}`
        );
      }

      columnIndexes = Object.fromEntries(
        args.columns.map((column) => [
          column,
          headers.indexOf(column),
        ])
      );

      continue;
    }

    for (const column of args.columns) {
      const index = columnIndexes[column];
      const exists =
        index !== -1 && index < fields.length;

      updateState(
        states[column],
        exists ? fields[index] : undefined,
        exists
      );
    }
  }

  if (headers === null) {
    throw new Error("No CSV header row found");
  }

  return states;
}

async function runJSONL(input, args) {
  const reader = readline.createInterface({
    input,
    crlfDelay: Infinity,
  });

  const states = createStates(
    args.columns,
    args.bitmapBits
  );

  let lineNumber = 0;

  for await (const line of reader) {
    lineNumber += 1;

    if (!line.trim()) {
      continue;
    }

    let record;

    try {
      record = JSON.parse(line);
    } catch (error) {
      throw new Error(
        `Invalid JSON on line ${lineNumber}: ${error.message}`
      );
    }

    if (
      record === null ||
      typeof record !== "object" ||
      Array.isArray(record)
    ) {
      throw new Error(
        `Invalid JSONL record on line ${lineNumber}: expected an object`
      );
    }

    for (const column of args.columns) {
      const result = getByPath(record, column);

      updateState(
        states[column],
        result.value,
        result.exists
      );
    }
  }

  return states;
}

function validateArgs(args) {
  if (!["csv", "jsonl"].includes(args.format)) {
    throw new Error(
      "--format must be csv or jsonl"
    );
  }

  if (!args.columns || args.columns.length === 0) {
    throw new Error(
      "--columns must contain at least one column"
    );
  }

  if (
    !Number.isInteger(args.bitmapBits) ||
    args.bitmapBits < 256
  ) {
    throw new Error(
      "--bitmap-bits must be an integer of at least 256"
    );
  }

  if (!args.delimiter) {
    throw new Error(
      "--delimiter cannot be empty"
    );
  }
}

async function main() {
  const args = parseArgs(process.argv);

  if (args.help) {
    usage();
    process.exit(0);
  }

  validateArgs(args);

  const input = args.file
    ? fs.createReadStream(args.file, {
        encoding: "utf8",
      })
    : process.stdin;

  const states =
    args.format === "csv"
      ? await runCSV(input, args)
      : await runJSONL(input, args);

  console.log(
    JSON.stringify(buildReport(states), null, 2)
  );
}

main().catch((error) => {
  console.error(
    "ERROR:",
    error && error.stack
      ? error.stack
      : error
  );

  process.exit(1);
});