AliveMCP

Guide · Data Serialization Integration

MCP Server Apache Parquet — read S3/GCS files via DuckDB, column projection, predicate pushdown

Three Apache Parquet behaviours trip up MCP server authors: reading Parquet with SELECT * scans all columns even though Parquet's columnar storage means you only pay for the columns you request — a table with 200 columns but an agent that only needs 3 will transfer 200 columns of data from S3 unless you write SELECT col1, col2, col3 explicitly; the fix is always to list the columns the tool response will use; DuckDB pushes WHERE predicates into Parquet row group statistics — a filter like WHERE date_col >= '2026-01-01' skips entire row groups whose max value is below the threshold without reading the data, but this only works if the file was written with row_group_size small enough to have useful min/max statistics (default 122,880 rows per row group is usually fine); and Hive-partitioned datasets require the hive_partitioning=true option — a path like s3://bucket/data/date=2026-07-01/*.parquet will not expose date as a queryable column unless you pass hive_partitioning=true to read_parquet(), which is the most effective partition pruning mechanism for date-ranged queries.

TL;DR

Use DuckDB's read_parquet() with HTTPFS extension to query Parquet files on S3 or GCS directly from MCP tool handlers. Always project columns explicitly — never use SELECT *. Pass hive_partitioning=true for partitioned datasets. Use parquet_schema() to inspect column types before building queries. Limit rows with LIMIT and use parquet_metadata() to understand file statistics before reading large datasets.

Reading Parquet from S3 via DuckDB HTTPFS

DuckDB's httpfs extension enables direct HTTP(S) reads from S3-compatible object stores and GCS without downloading files locally. The extension performs byte-range requests, so DuckDB can fetch only the column chunks and row groups it needs — combining with SELECT column projection and WHERE predicate pushdown makes remote Parquet queries practical even for files in the gigabyte range.

import Database from 'duckdb-async';

const db = await Database.create(':memory:');
const conn = await db.connect();

// Configure HTTPFS for S3 access — run once at startup
await conn.run(`
  INSTALL httpfs;
  LOAD httpfs;
  SET s3_region='${process.env.AWS_REGION ?? 'us-east-1'}';
  SET s3_access_key_id='${process.env.AWS_ACCESS_KEY_ID}';
  SET s3_secret_access_key='${process.env.AWS_SECRET_ACCESS_KEY}';
`);

// For GCS: use the S3-compatible endpoint
// SET s3_endpoint='storage.googleapis.com';
// SET s3_access_key_id=;
// SET s3_secret_access_key=;

// MCP tool: query a Parquet file with column projection and predicate pushdown
server.tool(
  'query_parquet',
  {
    s3_path:    z.string().startsWith('s3://'),
    columns:    z.array(z.string()).min(1).max(20),
    where_sql:  z.string().optional(),
    limit:      z.number().int().min(1).max(10000).default(1000),
    offset:     z.number().int().min(0).default(0),
  },
  async ({ s3_path, columns, where_sql, limit, offset }) => {
    // Sanitize column names — only allow identifier-safe names
    const safeCols = columns
      .map(c => c.replace(/[^a-zA-Z0-9_]/g, ''))
      .filter(Boolean)
      .map(c => `"${c}"`)
      .join(', ');

    if (!safeCols) throw new Error('No valid column names provided');

    // Build parameterised query (where_sql is validated separately)
    const whereClause = where_sql ? `WHERE ${where_sql}` : '';
    const sql = `
      SELECT ${safeCols}
      FROM read_parquet('${s3_path.replace(/'/g, "''")}'
        , hive_partitioning = true
        , union_by_name    = true   -- handle schema differences across files
      )
      ${whereClause}
      LIMIT ${limit}
      OFFSET ${offset}
    `;

    const rows = await conn.all(sql);
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ row_count: rows.length, rows }, (_, val) =>
          typeof val === 'bigint' ? val.toString() : val
        ),
      }],
    };
  }
);

// MCP tool: query a Hive-partitioned dataset with partition filter
server.tool(
  'query_partitioned_dataset',
  {
    s3_glob:    z.string(),  // e.g. 's3://my-bucket/events/**/*.parquet'
    date_from:  z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
    date_to:    z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
    columns:    z.array(z.string()).min(1).max(20),
    limit:      z.number().int().min(1).max(5000).default(500),
  },
  async ({ s3_glob, date_from, date_to, columns, limit }) => {
    const safeCols = columns.map(c => `"${c.replace(/[^a-zA-Z0-9_]/g, '')}"`).join(', ');

    // Partition filter on 'date' column — DuckDB will prune partitions
    // before fetching any data, not after
    const rows = await conn.all(`
      SELECT ${safeCols}
      FROM read_parquet('${s3_glob.replace(/'/g, "''")}', hive_partitioning = true)
      WHERE date >= '${date_from}' AND date <= '${date_to}'
      LIMIT ${limit}
    `);

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ row_count: rows.length, date_from, date_to, rows }, (_, val) =>
          typeof val === 'bigint' ? val.toString() : val
        ),
      }],
    };
  }
);

Inspecting Parquet schema and metadata before querying

DuckDB's parquet_schema() and parquet_metadata() table functions expose Parquet file metadata without reading row data. Use parquet_schema() to discover column names and types before building tool queries, and parquet_metadata() to understand row group count and row count to estimate whether a full scan is feasible within an MCP tool's time budget.

// MCP tool: inspect Parquet file schema — useful before building query tools
server.tool(
  'inspect_parquet_schema',
  { s3_path: z.string().startsWith('s3://') },
  async ({ s3_path }) => {
    const schema = await conn.all(`
      SELECT
        name         AS column_name,
        type         AS logical_type,
        converted_type,
        num_children
      FROM parquet_schema('${s3_path.replace(/'/g, "''")}')
      WHERE name != 'schema'  -- root schema node
      ORDER BY name
    `);

    const metadata = await conn.all(`
      SELECT
        COUNT(*)          AS row_group_count,
        SUM(num_values)   AS total_rows_approx,
        MIN(total_byte_size)  AS min_rg_bytes,
        MAX(total_byte_size)  AS max_rg_bytes,
        SUM(total_byte_size)  AS total_bytes
      FROM parquet_metadata('${s3_path.replace(/'/g, "''")}')
    `);

    const meta = metadata[0] as Record<string, unknown>;

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          columns: schema,
          stats: {
            row_groups:        meta.row_group_count,
            approx_total_rows: meta.total_rows_approx,
            total_size_bytes:  meta.total_bytes,
            total_size_mb:     Math.round(Number(meta.total_bytes) / 1_048_576 * 10) / 10,
          },
          recommendation:
            Number(meta.total_bytes) > 100_000_000
              ? 'File is >100 MB — always use column projection and WHERE predicates'
              : 'File is small enough for broad queries',
        }, (_, val) => typeof val === 'bigint' ? val.toString() : val),
      }],
    };
  }
);

// MCP tool: get row count without reading data (uses Parquet metadata footer)
server.tool(
  'count_parquet_rows',
  { s3_path: z.string() },
  async ({ s3_path }) => {
    // COUNT(*) on Parquet uses footer statistics — no row data fetched
    const result = await conn.all(`
      SELECT COUNT(*) AS row_count
      FROM read_parquet('${s3_path.replace(/'/g, "''")}')
    `);
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ row_count: (result[0] as { row_count: number }).row_count }),
      }],
    };
  }
);

Schema drift — handling Parquet files with mismatched columns

Parquet datasets written by different jobs or at different times often have slightly different schemas — a column may be added in a newer file or renamed. DuckDB's union_by_name=true option merges schemas across files in a glob pattern, filling missing columns with NULL. Without this option, reading a glob of files with different schemas throws a schema mismatch error. The trade-off is that union_by_name adds overhead for DuckDB to read all file footers before executing the query — for datasets with hundreds of files, this can add several seconds to query start time.

// Handle schema drift in multi-file datasets
server.tool(
  'query_evolving_dataset',
  {
    s3_glob:  z.string(),
    columns:  z.array(z.string()).min(1).max(20),
    limit:    z.number().int().min(1).max(1000).default(100),
  },
  async ({ s3_glob, columns, limit }) => {
    // First: discover union schema without reading rows
    const schema = await conn.all(`
      SELECT DISTINCT name, type
      FROM parquet_schema('${s3_glob.replace(/'/g, "''")}')
      WHERE name != 'schema'
      ORDER BY name
    `);

    // Validate requested columns exist in the union schema
    const knownColumns = new Set(schema.map((r: Record<string, unknown>) => r.name as string));
    const invalidCols = columns.filter(c => !knownColumns.has(c));
    if (invalidCols.length > 0) {
      return {
        content: [{
          type: 'text',
          text: JSON.stringify({
            error: `Columns not found in schema: ${invalidCols.join(', ')}`,
            available_columns: Array.from(knownColumns),
          }),
        }],
      };
    }

    const safeCols = columns.map(c => `"${c.replace(/[^a-zA-Z0-9_]/g, '')}"`).join(', ');
    const rows = await conn.all(`
      SELECT ${safeCols}
      FROM read_parquet('${s3_glob.replace(/'/g, "''")}',
        union_by_name    = true,
        hive_partitioning = true
      )
      LIMIT ${limit}
    `);

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ row_count: rows.length, rows }, (_, val) =>
          typeof val === 'bigint' ? val.toString() : val
        ),
      }],
    };
  }
);

Health probe — verify DuckDB HTTPFS and S3 access at startup

An MCP server that reads Parquet from S3 can fail for multiple independent reasons: the HTTPFS extension is not installed, the S3 credentials are expired, the bucket policy changed, or the target file was deleted. Separate the probe into two checks — one for HTTPFS loading and one for an actual S3 connectivity test using a known small reference file. This makes it easy to diagnose which layer failed when AliveMCP reports the endpoint as unhealthy.

async function checkParquetHealth(): Promise<{
  httpfs_loaded: boolean;
  s3_accessible: boolean;
  probe_rows: number;
  latency_ms: number;
  error?: string;
}> {
  const start = Date.now();

  // Check 1: HTTPFS extension loaded
  let httpfsLoaded = false;
  try {
    const extensions = await conn.all(`
      SELECT extension_name, loaded FROM duckdb_extensions()
      WHERE extension_name = 'httpfs'
    `);
    httpfsLoaded = extensions.some((e: Record<string, unknown>) => e.loaded === true);
  } catch {
    return { httpfs_loaded: false, s3_accessible: false, probe_rows: 0, latency_ms: Date.now() - start, error: 'duckdb_extensions() query failed' };
  }

  if (!httpfsLoaded) {
    return { httpfs_loaded: false, s3_accessible: false, probe_rows: 0, latency_ms: Date.now() - start, error: 'httpfs extension not loaded' };
  }

  // Check 2: S3 connectivity — read 1 row from a known small probe file
  const probeFile = process.env.PARQUET_HEALTH_PROBE_FILE; // e.g. 's3://my-bucket/probe/health.parquet'
  if (!probeFile) {
    return { httpfs_loaded: true, s3_accessible: false, probe_rows: 0, latency_ms: Date.now() - start, error: 'PARQUET_HEALTH_PROBE_FILE not set' };
  }

  try {
    const rows = await conn.all(`
      SELECT COUNT(*) AS n FROM read_parquet('${probeFile}') LIMIT 1
    `);
    return {
      httpfs_loaded: true,
      s3_accessible: true,
      probe_rows: (rows[0] as { n: number }).n,
      latency_ms: Date.now() - start,
    };
  } catch (err) {
    return {
      httpfs_loaded: true,
      s3_accessible: false,
      probe_rows: 0,
      latency_ms: Date.now() - start,
      error: String(err),
    };
  }
}

server.resource('parquet_health', 'parquet://health', async () => {
  const health = await checkParquetHealth();
  return {
    contents: [{
      uri: 'parquet://health',
      text: JSON.stringify(health),
    }],
  };
});