Build a Node.js backend that analyzes hundreds of millions of Parquet and CSV records with SQL — no separate server required

pnpm add @duckdb/node-api

// src/analytics/duckdb.client.ts
import { DuckDBInstance } from '@duckdb/node-api';
 
async function quickStart() {
  // ':memory:' — in-memory DB, data is lost when the process exits
  const instance = await DuckDBInstance.create(':memory:');
  const conn = await instance.connect();
 
  const result = await conn.run(`
    SELECT 42 AS answer, 'DuckDB works!' AS message
  `);
 
  const rows = await result.fetchAllRows();
  console.log(rows);
  // [{ answer: 42, message: 'DuckDB works!' }]
 
  await conn.close();
  await instance.close();
}
 
quickStart();

const instance = await DuckDBInstance.create('./analytics.duckdb');

// src/analytics/path-validator.ts
import path from 'node:path';
 
export const DATA_BASE_DIR = path.resolve(process.env.DATA_BASE_DIR ?? './data');
 
export function validateDataPath(input: string): string {
  // Reject any characters other than alphanumerics, slashes, hyphens, underscores, dots, and globs (*)
  // Single quotes (') and other SQL special characters are blocked at this stage
  if (!/^[\w\-./\*]+$/.test(input)) {
    throw new Error('Path contains disallowed characters');
  }
  const resolved = path.resolve(DATA_BASE_DIR, input);
  if (!resolved.startsWith(DATA_BASE_DIR + path.sep) && resolved !== DATA_BASE_DIR) {
    throw new Error('Path is outside the allowed directory');
  }
  return resolved;
}

// src/analytics/analytics.service.ts
import { validateDataPath } from './path-validator';
 
async getSalesSummary(year: number) {
  // Even for number types, validate that it's an integer within range
  if (!Number.isInteger(year) || year < 2000 || year > 2100) {
    throw new Error('Invalid year');
  }
 
  // Query multiple Parquet files at once with a glob pattern
  const safePath = validateDataPath(`sales/${year}/*.parquet`);
 
  const result = await this.conn.run(`
    SELECT
      region,
      SUM(amount)   AS total_sales,
      COUNT(*)      AS order_count,
      AVG(amount)   AS avg_order_value
    FROM '${safePath}'
    WHERE status = 'completed'
    GROUP BY region
    ORDER BY total_sales DESC
  `);
 
  return result.fetchAllRows();
}
 
async analyzeCSV(relativePath: string) {
  const safePath = validateDataPath(relativePath);
 
  const result = await this.conn.run(`
    SELECT *
    FROM read_csv_auto('${safePath}', header = true)
    LIMIT 100
  `);
 
  return result.fetchAllRows();
}

async getTopProducts(categoryId: number, limit: number) {
  const safePath = validateDataPath('orders/*.parquet');
 
  const stmt = await this.conn.prepare(`
    SELECT
      product_id,
      product_name,
      SUM(quantity) AS total_sold
    FROM '${safePath}'
    WHERE category_id = $categoryId
    GROUP BY product_id, product_name
    ORDER BY total_sold DESC
    LIMIT $limit
  `);
 
  try {
    const result = await stmt.run({ categoryId, limit });
    return await result.fetchAllRows();
  } finally {
    await stmt.close();
  }
}

// src/analytics/duckdb.provider.ts
import { DuckDBInstance, DuckDBConnection } from '@duckdb/node-api';
import { ConfigService } from '@nestjs/config';
 
export const DUCKDB_INSTANCE = 'DUCKDB_INSTANCE';
export const DUCKDB_CONNECTION = 'DUCKDB_CONNECTION';
 
export const DuckDBProviders = [
  {
    provide: DUCKDB_INSTANCE,
    useFactory: async (config: ConfigService): Promise<DuckDBInstance> => {
      const dbPath = config.get<string>('DUCKDB_PATH', ':memory:');
      const instance = await DuckDBInstance.create(dbPath);
 
      // threads and memory_limit are configured via SET commands
      // The second-argument signature of DuckDBInstance.create can vary across versions,
      // so standard SQL SET commands are the safest and most version-independent approach
      const configConn = await instance.connect();
      try {
        const threads = parseInt(config.get<string>('DUCKDB_THREADS', '4'), 10);
        const memLimit = config.get<string>('DUCKDB_MEMORY_LIMIT', '2GB');
        await configConn.run(`SET threads = ${threads}`);
        await configConn.run(`SET memory_limit = '${memLimit}'`);
      } finally {
        await configConn.close();
      }
 
      return instance;
    },
    inject: [ConfigService],
  },
  {
    provide: DUCKDB_CONNECTION,
    useFactory: async (instance: DuckDBInstance): Promise<DuckDBConnection> => {
      return instance.connect();
    },
    inject: [DUCKDB_INSTANCE],
  },
];

// src/analytics/analytics.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { DuckDBProviders } from './duckdb.provider';
import { AnalyticsService } from './analytics.service';
import { AnalyticsController } from './analytics.controller';
 
@Module({
  imports: [ConfigModule],
  providers: [...DuckDBProviders, AnalyticsService],
  controllers: [AnalyticsController],
  exports: [AnalyticsService],
})
export class AnalyticsModule {}

// src/analytics/analytics.service.ts
import { Injectable, Inject, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { DuckDBInstance, DuckDBConnection } from '@duckdb/node-api';
import { DUCKDB_INSTANCE, DUCKDB_CONNECTION } from './duckdb.provider';
import { validateDataPath } from './path-validator';
 
@Injectable()
export class AnalyticsService implements OnModuleInit, OnModuleDestroy {
  constructor(
    @Inject(DUCKDB_INSTANCE) private readonly instance: DuckDBInstance,
    @Inject(DUCKDB_CONNECTION) private readonly conn: DuckDBConnection,
  ) {}
 
  async onModuleInit() {
    // Load the httpfs extension once during module initialization
    // INSTALL only performs an actual download on first run; subsequent calls are no-ops
    await this.conn.run(`INSTALL httpfs; LOAD httpfs;`);
 
    // S3 credentials: CREDENTIAL_CHAIN searches environment variables, IAM Roles,
    // and instance profiles in order, so no keys need to be exposed in the code
    await this.conn.run(`
      CREATE OR REPLACE SECRET aws_s3 (
        TYPE S3,
        PROVIDER CREDENTIAL_CHAIN
      );
    `);
  }
 
  async onModuleDestroy() {
    try {
      await this.conn.close();
    } finally {
      await this.instance.close();
    }
  }
 
  async getDailySummary(date: string) {
    // Validate YYYY-MM-DD format
    if (!/^\d{4}-\d{2}-\d{2}$/.test(date)) {
      throw new Error('Invalid date format');
    }
 
    const safePath = validateDataPath(`events/${date}/*.parquet`);
    const stmt = await this.conn.prepare(`
      SELECT
        DATE_TRUNC('hour', event_time) AS hour,
        event_type,
        COUNT(*) AS event_count
      FROM '${safePath}'
      WHERE event_time >= $startDate::TIMESTAMP
        AND event_time < $startDate::TIMESTAMP + INTERVAL 1 DAY
      GROUP BY 1, 2
      ORDER BY 1, 2
    `);
 
    try {
      const result = await stmt.run({ startDate: date });
      return await result.fetchAllRows();
    } finally {
      await stmt.close();
    }
  }
}

async queryS3Parquet(s3Prefix: string) {
  // s3Prefix example: 'logs/2026/07'
  // Recommend validating against an allowed prefix whitelist for S3 paths
  if (!/^[\w\-./]+$/.test(s3Prefix)) {
    throw new Error('Invalid S3 path');
  }
 
  const result = await this.conn.run(`
    SELECT
      user_id,
      COUNT(*) AS page_views,
      COUNT(DISTINCT session_id) AS sessions
    FROM 's3://my-analytics-bucket/${s3Prefix}/*.parquet'
    WHERE event_type = 'page_view'
    GROUP BY user_id
    HAVING page_views > 100
    ORDER BY page_views DESC
    LIMIT 1000
  `);
 
  return result.fetchAllRows();
}

async *streamResultByPage(
  relativePath: string,
  pageSize = 10_000,
): AsyncGenerator<Record<string, unknown>[]> {
  const safePath = validateDataPath(relativePath);
  let offset = 0;
 
  while (true) {
    const stmt = await this.conn.prepare(`
      SELECT * FROM '${safePath}'
      LIMIT $pageSize OFFSET $offset
    `);
 
    try {
      const result = await stmt.run({ pageSize, offset });
      const rows = await result.fetchAllRows() as Record<string, unknown>[];
      if (rows.length === 0) break;
      yield rows;
      if (rows.length < pageSize) break;
      offset += pageSize;
    } finally {
      await stmt.close();
    }
  }
}

for await (const page of service.streamResultByPage('large-dataset.parquet')) {
  await processBatch(page);
}

async convertCSVToParquet(csvRelPath: string, outRelPath: string) {
  const csvPath = validateDataPath(csvRelPath);
  const outPath = validateDataPath(outRelPath);
 
  await this.conn.run(`
    COPY (
      SELECT * FROM read_csv_auto('${csvPath}', header = true)
    ) TO '${outPath}'
    (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000)
  `);
}

// Dangerous — blocks neither path traversal nor SQL special characters
await conn.run(`SELECT * FROM '${userInput}'`);
 
// Wrong approach — path.basename only prevents directory traversal;
// if userInput contains a ' character, the SQL is still broken
const name = path.basename(userInput);
await conn.run(`SELECT * FROM './data/${name}'`);
 
// Correct approach — validateDataPath validates both allowed characters and directory boundaries simultaneously
const safePath = validateDataPath(userInput);
await conn.run(`SELECT * FROM '${safePath}'`);