モダンWebアプリにおけるDuckDB-Wasm：クライアントサイドOLAP、Parquetストリーミング、そして超高速ダッシュボード

// duckdb.worker.ts
import * as duckdb from '@duckdb/duckdb-wasm';
import duckdb_wasm from '@duckdb/duckdb-wasm/dist/duckdb-mvp.wasm';
import duckdb_wasm_next from '@duckdb/duckdb-wasm/dist/duckdb-eh.wasm';

// Define the DuckDB bundle configuration
const DUCKDB_BUNDLES: duckdb.DuckDBBundles = {
  mvp: {
    mainModule: duckdb_wasm,
    mainWorker: new URL('@duckdb/duckdb-wasm/dist/duckdb-browser-mvp.worker.js', import.meta.url).toString(),
  },
  eh: {
    mainModule: duckdb_wasm_next,
    mainWorker: new URL('@duckdb/duckdb-wasm/dist/duckdb-browser-eh.worker.js', import.meta.url).toString(),
  },
};

let db: duckdb.AsyncDuckDB | null = null;
let conn: duckdb.AsyncDuckDBConnection | null = null;

// Initialize DuckDB and establish a connection
async function initializeDuckDB() {
  if (db && conn) return;

  const logger = new duckdb.ConsoleLogger();
  const bundle = await duckdb.selectBundle(DUCKDB_BUNDLES);

  // Instantiate the database
  db = new duckdb.AsyncDuckDB(logger, bundle);
  await db.instantiate(bundle.mainWorker);
  conn = await db.connect();

  console.log('DuckDB-Wasm initialized in worker.');
}

// Handle messages from the main thread
self.onmessage = async (event: MessageEvent) => {
  const { id, type, payload } = event.data;

  try {
    if (type === 'init') {
      await initializeDuckDB();
      self.postMessage({ id, type: 'init_success' });
    } else if (type === 'execute_sql') {
      if (!conn) throw new Error('DuckDB connection not established.');

      const { sql } = payload;
      console.log(`Executing SQL: ${sql}`);

      // Execute query and get results as Arrow
      const result = await conn.query(sql);

      // Transfer Arrow IPC buffer back to main thread
      // The toIPC method serializes the Arrow table into an IPC stream buffer.
      const arrowBuffer = result.toIPC();
      self.postMessage({ id, type: 'execute_sql_success', payload: arrowBuffer }, [arrowBuffer]);
    } else if (type === 'register_parquet_url') {
      if (!db) throw new Error('DuckDB not initialized.');

      const { url, tableName } = payload;
      console.log(`Registering Parquet URL: ${url} as ${tableName}`);

      // Register a remote Parquet file. DuckDB-Wasm handles HTTP range requests internally.
      await db.registerFileURL(tableName, url, duckdb.DuckDBDataProtocol.HTTP, false);
      self.postMessage({ id, type: 'register_parquet_url_success' });
    } else {
      throw new Error(`Unknown message type: ${type}`);
    }
  } catch (error: any) {
    console.error(`Worker error for message ID ${id}:`, error);
    self.postMessage({ id, type: 'error', payload: error.message });
  }
};

// duckdb.service.ts
import { Table } from 'apache-arrow';

// Using a dedicated worker for DuckDB operations
const worker = new Worker(new URL('./duckdb.worker.ts', import.meta.url), { type: 'module' });

// Map to store pending requests and resolve them
const pendingRequests = new Map<string, { resolve: Function; reject: Function }>();
let messageIdCounter = 0;

worker.onmessage = (event: MessageEvent) => {
  const { id, type, payload } = event.data;
  const request = pendingRequests.get(id);

  if (request) {
    if (type === 'error') {
      request.reject(new Error(payload));
    } else if (type === 'execute_sql_success') {
      // Reconstruct Arrow Table from the transferred buffer
      const table = Table.from([new Uint8Array(payload)]);
      request.resolve(table);
    } else {
      request.resolve(payload);
    }
    pendingRequests.delete(id);
  } else {
    console.warn(`Received message for unknown request ID: ${id}`);
  }
};

function sendMessageToWorker(type: string, payload?: any): Promise<any> {
  return new Promise((resolve, reject) => {
    const id = `msg_${messageIdCounter++}`;
    pendingRequests.set(id, { resolve, reject });
    worker.postMessage({ id, type, payload });
  });
}

export const duckdbService = {
  init: () => sendMessageToWorker('init'),

  /**
   * Registers a remote Parquet file URL with DuckDB.
   * DuckDB-Wasm will use HTTP range requests to access this file.
   * @param url The URL of the Parquet file.
   * @param tableName The name to register the table as in DuckDB.
   */
  registerParquetUrl: (url: string, tableName: string) =>
    sendMessageToWorker('register_parquet_url', { url, tableName }),

  /**
   * Executes a SQL query and returns the result as an Apache Arrow Table.
   * @param sql The SQL query string.
   * @returns A Promise that resolves to an Apache Arrow Table.
   */
  executeSql: async (sql: string): Promise<Table> =>
    sendMessageToWorker('execute_sql', { sql }),
};

// Initialize DuckDB-Wasm when the service is imported
duckdbService.init().catch(console.error);

// Example usage in a React component or similar
import React, { useEffect, useState } from 'react';
import { duckdbService } from './duckdb.service';
import { Table } from 'apache-arrow';

const ParquetDataLoader: React.FC = () => {
  const [data, setData] = useState<Table | null>(null);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    const loadData = async () => {
      try {
        // Ensure DuckDB is initialized
        await duckdbService.init();

        const parquetUrl = 'https://example.com/path/to/your/large_dataset.parquet';
        const tableName = 'my_data';

        // Register the remote Parquet file
        await duckdbService.registerParquetUrl(parquetUrl, tableName);

        // Execute a query to fetch some data
        // DuckDB will automatically fetch only the necessary parts of the Parquet file
        const resultTable = await duckdbService.executeSql(`
          SELECT
            category,
            SUM(value) AS total_value,
            COUNT(*) AS record_count
          FROM my_data
          WHERE timestamp > '2023-01-01'
          GROUP BY category
          ORDER BY total_value DESC
          LIMIT 10;
        `);

        setData(resultTable);
      } catch (err: any) {
        setError(err.message);
        console.error('Failed to load Parquet data:', err);
      } finally {
        setLoading(false);
      }
    };

    loadData();
  }, []);

  if (loading) return <div>Loading data...</div>;
  if (error) return <div>Error: {error}</div>;
  if (!data) return <div>No data loaded.</div>;

  return (
    <div>
      <h2>Top Categories by Value</h2>
      <pre>{JSON.stringify(data.toArray().map(row => row.toJSON()), null, 2)}</pre>
      {/* Render chart here using 'data' */}
    </div>
  );
};

export default ParquetDataLoader;

// components/ArrowScatterPlot.tsx
import React, { useRef, useEffect } from 'react';
import { Table } from 'apache-arrow';

interface ArrowScatterPlotProps {
  data: Table;
  xColumn: string;
  yColumn: string;
  width?: number;
  height?: number;
  pointColor?: string;
  pointRadius?: number;
}

const ArrowScatterPlot: React.FC<ArrowScatterPlotProps> = ({
  data,
  xColumn,
  yColumn,
  width = 800,
  height = 400,
  pointColor = 'rgba(75, 192, 192, 0.5)',
  pointRadius = 2,
}) => {
  const canvasRef = useRef<HTMLCanvasElement>(null);

  useEffect(() => {
    if (!data || !canvasRef.current) return;

    const canvas = canvasRef.current;
    const ctx = canvas.getContext('2d');
    if (!ctx) return;

    ctx.clearRect(0, 0, width, height); // Clear previous drawing

    const xData = data.getChild(xColumn);
    const yData = data.getChild(yColumn);

    if (!xData || !yData || xData.length !== yData.length) {
      console.warn('Invalid x or y column data for scatter plot.');
      return;
    }

    // Determine data bounds for scaling
    const minX = Math.min(...(xData.toArray() as number[]));
    const maxX = Math.max(...(xData.toArray() as number[]));
    const minY = Math.min(...(yData.toArray() as number[]));
    const maxY = Math.max(...(yData.toArray() as number[]));

    // Simple linear scaling functions
    const scaleX = (value: number) => {
      if (maxX === minX) return width / 2; // Avoid division by zero
      return ((value - minX) / (maxX - minX)) * width;
    };
    const scaleY = (value: number) => {
      if (maxY === minY) return height / 2; // Avoid division by zero
      // Invert Y-axis for canvas coordinates (0,0 is top-left)
      return height - (((value - minY) / (maxY - minY)) * height);
    };

    ctx.fillStyle = pointColor;
    ctx.beginPath();

    // Iterate directly over Arrow vectors for performance
    // Using get() is slower than iterating over underlying TypedArrays if possible,
    // but for demonstration, get() is simpler. For extreme performance,
    // access xData.data.values directly if it's a primitive type.
    for (let i = 0; i < data.numRows; i++) {
      const x = xData.get(i) as number;
      const y = yData.get(i) as number;

      if (x !== null && y !== null && !isNaN(x) && !isNaN(y)) {
        const screenX = scaleX(x);
        const screenY = scaleY(y);

        ctx.moveTo(screenX, screenY); // Move before arc to avoid connecting dots
        ctx.arc(screenX, screenY, pointRadius, 0, Math.PI * 2);
      }
    }
    ctx.fill();

    // Optional: Draw axes and labels
    ctx.strokeStyle = '#ccc';
    ctx.lineWidth = 1;
    ctx.font = '10px Arial';
    ctx.fillStyle = '#333';

    // X-axis
    ctx.beginPath();
    ctx.moveTo(0, height);
    ctx.lineTo(width, height);
    ctx.stroke();
    ctx.fillText(minX.toFixed(2), 0, height - 5);
    ctx.fillText(maxX.toFixed(2), width - 30, height - 5);

    // Y-axis
    ctx.beginPath();
    ctx.moveTo(0, 0);
    ctx.lineTo(0, height);
    ctx.stroke();
    ctx.fillText(maxY.toFixed(2), 5, 15);
    ctx.fillText(minY.toFixed(2), 5, height - 5);

  }, [data, xColumn, yColumn, width, height, pointColor, pointRadius]);

  return <canvas ref={canvasRef} width={width} height={height} style={{ border: '1px solid #eee' }} />;
};

export default ArrowScatterPlot;