CSV to SQL Script Generator

CSV to SQL: Schema Inference, COPY vs INSERT, Null Handling, and PostgreSQL/MySQL Dialects

CSV-to-SQL conversion has three hard problems beyond wrapping rows in INSERT statements: schema inference (how do you distinguish an integer column from a string column that happens to contain only numbers?), NULL semantics (is an empty CSV cell a NULL or an empty string?), and bulk loading performance (individual INSERTs at 100 rows/second vs PostgreSQL COPY at 100,000+ rows/second). This guide covers a reliable type inference algorithm, NULL handling strategies, PostgreSQL's COPY command, MySQL's LOAD DATA INFILE, transaction batching for large files, and a staging table pattern for safe production imports.

Live Example: Schema Inference and SQL Generation

import Papa from 'papaparse';

type SqlType = 'INTEGER' | 'NUMERIC(15,4)' | 'BOOLEAN' | 'DATE' | 'TIMESTAMP' | 'TEXT';

// Sample N rows to infer column types
function inferSqlType(values: string[]): SqlType {
  const nonEmpty = values.filter(v => v !== '' && v.toLowerCase() !== 'null');
  if (nonEmpty.length === 0) return 'TEXT';  // all empty — default to TEXT

  // Test each candidate type — most restrictive first
  if (nonEmpty.every(v => /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}/.test(v)))  return 'TIMESTAMP';
  if (nonEmpty.every(v => /^\d{4}-\d{2}-\d{2}$/.test(v)))                     return 'DATE';
  if (nonEmpty.every(v => /^(true|false|yes|no|1|0)$/i.test(v)))              return 'BOOLEAN';
  if (nonEmpty.every(v => /^-?\d+$/.test(v)))                                 return 'INTEGER';
  if (nonEmpty.every(v => /^-?\d+(\.\d+)?$/.test(v)))                         return 'NUMERIC(15,4)';
  return 'TEXT';
}

function csvToCreateTable(
  csvString: string,
  tableName: string,
  dialect: 'postgresql' | 'mysql' | 'sqlite' = 'postgresql'
): string {
  const { data, meta } = Papa.parse(csvString, {
    header: true,
    skipEmptyLines: true,
    preview: 200,  // sample first 200 rows for type inference
  });

  const headers  = meta.fields ?? [];
  const rows     = data as Record<string, string>[];

  const columnDefs = headers.map(col => {
    const values   = rows.map(r => r[col] ?? '');
    const sqlType  = inferSqlType(values);
    const nullable = values.some(v => v === '' || v.toLowerCase() === 'null');
    const notNull  = nullable ? '' : ' NOT NULL';

    // Sanitize column name (remove spaces, special chars)
    const safeName = col.trim().toLowerCase().replace(/\W+/g, '_');
    const quoted   = dialect === 'mysql' ? `\`${safeName}\`` : `"${safeName}"`;

    return `  ${quoted} ${sqlType}${notNull}`;
  });

  const autoId = dialect === 'postgresql'
    ? '  id BIGSERIAL PRIMARY KEY,'
    : '  id BIGINT AUTO_INCREMENT PRIMARY KEY,';

  return `CREATE TABLE ${dialect === 'mysql' ? `\`${tableName}\`` : `"${tableName}"`} (\n${autoId}\n${columnDefs.join(',\n')}\n);`;
}

// Usage
const createSQL = csvToCreateTable(csvString, 'transactions', 'postgresql');
// CREATE TABLE "transactions" (
//   id BIGSERIAL PRIMARY KEY,
//   "transaction_id" TEXT NOT NULL,
//   "amount" NUMERIC(15,4) NOT NULL,
//   "occurred_at" TIMESTAMP NOT NULL,
//   "description" TEXT,
//   "is_refund" BOOLEAN NOT NULL
// );

PostgreSQL COPY: Bulk Loading

// COPY is 50-100x faster than individual INSERTs for large CSV files
// Requires direct file access on the server OR COPY FROM STDIN via the client

// Option 1: Server-side COPY (requires superuser or file access)
COPY transactions ("transaction_id", "amount", "occurred_at", "description", "is_refund")
FROM '/tmp/transactions.csv'
WITH (
  FORMAT CSV,
  HEADER TRUE,
  DELIMITER ',',
  NULL '',          -- empty field = NULL
  ENCODING 'UTF8',
  QUOTE '"',
  ESCAPE '"'
);

// Option 2: COPY FROM STDIN via pg (no file access needed)
// npm install pg
import { Client } from 'pg';
import { createReadStream } from 'fs';
import { pipeline } from 'stream/promises';

async function copyCsvToPostgres(csvPath: string, tableName: string, columns: string[]): Promise<void> {
  const client = new Client({ connectionString: process.env.DATABASE_URL });
  await client.connect();

  const colList  = columns.map(c => `"${c}"`).join(', ');
  const copySQL  = `COPY "${tableName}" (${colList}) FROM STDIN WITH (FORMAT CSV, HEADER TRUE, NULL '')`;
  const stream   = client.query(require('pg-copy-streams').from(copySQL));

  await pipeline(createReadStream(csvPath), stream);
  await client.end();
  console.log(`Loaded ${stream.rowCount} rows`);
}

await copyCsvToPostgres('./transactions.csv', 'transactions', [
  'transaction_id', 'amount', 'occurred_at', 'description', 'is_refund',
]);

MySQL LOAD DATA and Batched INSERTs

-- MySQL bulk load (requires FILE privilege or LOCAL keyword)
LOAD DATA LOCAL INFILE '/tmp/transactions.csv'
INTO TABLE `transactions`
FIELDS TERMINATED BY ','
OPTIONALLY ENCLOSED BY '"'
LINES TERMINATED BY '\n'
IGNORE 1 ROWS           -- skip header
("transaction_id", "amount", @occurred_at, "description", @is_refund)
SET occurred_at = STR_TO_DATE(@occurred_at, '%Y-%m-%dT%H:%i:%sZ'),
    is_refund   = IF(@is_refund = 'true', TRUE, FALSE);

// For environments without FILE privilege, use batched INSERTs:
// Generates: INSERT INTO table (col1, col2) VALUES (v1, v2), (v3, v4), ... (1000 rows per batch)

function generateBatchInserts(
  rows: Record<string, string>[],
  tableName: string,
  columns: string[],
  batchSize = 1000
): string[] {
  const sqls: string[] = [];
  const colList = columns.map(c => `\`${c}\``).join(', ');

  for (let i = 0; i < rows.length; i += batchSize) {
    const batch = rows.slice(i, i + batchSize);
    const values = batch.map(row =>
      '(' + columns.map(col => {
        const v = row[col] ?? '';
        if (v === '' || v.toLowerCase() === 'null') return 'NULL';
        if (/^-?\d+(\.\d+)?$/.test(v)) return v;           // numeric — no quotes
        if (/^(true|false)$/i.test(v)) return v.toUpperCase();  // boolean
        return `'${v.replace(/'/g, "''")}'`;                // string — escape quotes
      }).join(', ') + ')'
    ).join(',\n  ');

    sqls.push(`INSERT INTO \`${tableName}\` (${colList}) VALUES\n  ${values};`);
  }

  return sqls;
}

Staging Table Pattern for Safe Production Imports

-- Safe import workflow:
-- 1. Load into staging table (no constraints)
-- 2. Validate and clean
-- 3. Move clean rows to production table
-- 4. Report or quarantine invalid rows

-- Step 1: Create staging table (all TEXT, no constraints)
CREATE TABLE staging_transactions (
  transaction_id TEXT,
  amount         TEXT,
  occurred_at    TEXT,
  description    TEXT,
  is_refund      TEXT
);

-- Load CSV (no type errors possible — everything is TEXT)
COPY staging_transactions FROM STDIN WITH (FORMAT CSV, HEADER TRUE);

-- Step 2: Validate
CREATE TABLE import_errors AS
SELECT *, 'Invalid amount' AS error_reason
FROM staging_transactions
WHERE amount ~ '[^0-9.-]' OR amount IS NULL
UNION ALL
SELECT *, 'Invalid timestamp'
FROM staging_transactions
WHERE occurred_at !~ '^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}';

-- Step 3: Insert valid rows with type casting
INSERT INTO transactions (transaction_id, amount, occurred_at, description, is_refund)
SELECT
  transaction_id,
  amount::NUMERIC(15,4),
  occurred_at::TIMESTAMP,
  NULLIF(description, ''),
  is_refund = 'true'
FROM staging_transactions
WHERE transaction_id NOT IN (SELECT transaction_id FROM import_errors);

-- Step 4: Report
SELECT COUNT(*) AS total, COUNT(*) FILTER (WHERE transaction_id IN (SELECT transaction_id FROM import_errors)) AS errors
FROM staging_transactions;

-- Cleanup
DROP TABLE staging_transactions;

Best Practices for Production

• Use PostgreSQL COPY or MySQL LOAD DATA for files over 10,000 rows: Individual INSERTs parse, plan, and execute each statement independently. COPY bypasses the query planner for each row, operating at ~1M rows/minute on modern hardware vs ~1,000 rows/minute for individual INSERTs.
• Define NULL semantics before importing: In CSV, an empty field (,,) could mean NULL, empty string, or zero. Decide upfront. PostgreSQL COPY's NULL '' option treats empty fields as NULL. An explicit NULL text value requires NULL 'NULL'. Without clarity, you'll have mixed semantics in your imported data.
• Use a staging table for any production import: Loading directly to a production table with constraints means one bad row aborts the entire import. Load to a staging table with no constraints, validate, report bad rows, and move clean data. This separates data quality from data loading.
• Wrap INSERT statements in transactions for atomicity: For non-COPY approaches, wrap each batch in BEGIN; ... COMMIT;. If any INSERT fails, the transaction rolls back that batch cleanly. Without transactions, partial imports leave the table in a mixed state.

FAQ

Q: How do I handle CSV files with encoding issues (Latin-1, Windows-1252)?
A: Detect encoding with the chardet library (Python) or chardet npm package, convert to UTF-8 with iconv before importing. PostgreSQL COPY supports ENCODING 'LATIN1' directly. MySQL LOAD DATA has CHARACTER SET latin1. Never assume UTF-8 for files exported from Windows Excel — they're frequently Latin-1 or Windows-1252.

Q: How do I handle duplicate rows during import?
A: PostgreSQL: INSERT ... ON CONFLICT (unique_col) DO NOTHING or DO UPDATE SET ... (upsert). MySQL: INSERT IGNORE or ON DUPLICATE KEY UPDATE. For COPY, there's no native conflict handling — use the staging table pattern and deduplicate with SELECT DISTINCT before inserting.

Q: What's the fastest way to import a 1GB CSV into PostgreSQL?
A: Use COPY with FORMAT BINARY if you can convert the CSV to binary format first. Otherwise, plain COPY FORMAT CSV achieves 100-500 MB/min. Disable indexes before bulk import and rebuild them after: DROP INDEX idx_...; COPY ...; CREATE INDEX CONCURRENTLY idx_ .... Also set synchronous_commit = off for the session.