docs/data-layer.md

Data Loading and Schema

This document describes the data loading pipeline, from CSV upload through schema inference and statistics computation, as well as data export capabilities.

CSV Loading Pipeline

The CSV loading pipeline consists of four stages: file upload, disk persistence, DuckDB registration, and preview generation. Each stage is designed to handle files up to the configured maximum upload size (default 500 MB).

Stage 1: File Upload and Validation

When a user uploads a CSV file through POST /api/upload/csv, the endpoint performs the following validation steps:

1. Filename check: The file must have a .csv extension. Non-CSV files are rejected with a 400 error.
2. Size check: The uploaded content is read into memory and compared against MAX_UPLOAD_SIZE. Files exceeding this limit receive a 413 error.
3. Sanitization: The filename is sanitized by replacing spaces and forward slashes with underscores to prevent path traversal issues.

if not file.filename.endswith(".csv"):
    raise HTTPException(status_code=400, detail="Only CSV files are supported")
content = await file.read()
if len(content) > settings.max_upload_size:
    raise HTTPException(status_code=413, detail="File too large")

Stage 2: Disk Persistence

The uploaded file is saved to the DATA_DIR directory with a unique prefix to avoid collisions:

dataset_id = uuid.uuid4().hex[:8]
save_path = Path(settings.data_dir) / f"{dataset_id}_{safe_name}"
with open(save_path, "wb") as f:
    f.write(content)

The dataset_id is an 8-character hex string derived from a UUID4, providing sufficient uniqueness for session-scoped usage. The original filename is preserved after the prefix for user identification.

Stage 3: DuckDB Registration

The register_csv() function in csv_loader.py loads the CSV file into DuckDB using the read_csv_auto function:

con.execute(
    f'CREATE OR REPLACE TABLE "{table_name}" AS SELECT * FROM read_csv_auto(\'{csv_path}\', header=true, ignore_errors=true)'
)

Key aspects of this approach:

• read_csv_auto: DuckDB's automatic CSV reader detects delimiters, quote characters, escape characters, and column types without manual configuration. This eliminates the need for the user to specify CSV dialect parameters.
• ignore_errors=true: Malformed rows are silently skipped rather than causing the entire load to fail. This is important for real-world CSV files that may contain inconsistent quoting or type mismatches.
• CREATE OR REPLACE TABLE: The data is materialized into a named table in the DuckDB in-memory database, making it available for subsequent SQL queries.

After registration, the function queries the row count and column information:

row_count = con.execute(f'SELECT COUNT(*) FROM "{table_name}"').fetchone()[0]
col_info = con.execute(
    f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{table_name}'"
).fetchall()

Stage 4: Polars Sampling for Fast Preview

For large files, full table scans can be expensive. The read_csv_sample() function provides a fast preview by reading only the first N rows using Polars:

def read_csv_sample(path: str, n: int = 10_000) -> pl.DataFrame:
    return pl.read_csv(path, n_rows=n, ignore_errors=True)

The n_rows parameter in Polars reads only the specified number of rows from disk, avoiding the overhead of loading the entire file. This is useful for UI previews where exact results are not required.

The read_csv_lazy() function provides a LazyFrame interface for deferred computation:

def read_csv_lazy(path: str) -> pl.LazyFrame:
    return pl.scan_csv(path, ignore_errors=True)

LazyFrames enable query optimization and allow the Polars query planner to push predicates and projections down to the scan level, reducing memory usage and computation time.

Schema Inference

Primary Method: DuckDB SUMMARIZE

The primary schema inference method uses DuckDB's SUMMARIZE command, which computes comprehensive statistics for every column in a single pass:

rows = con.execute(f'SUMMARIZE "{table_name}"').fetchall()

The SUMMARIZE command returns the following columns for each table column:

SUMMARIZE Column	Description	Maps To
column_name	Column identifier	name
column_type	DuckDB data type	dtype
min	Minimum value	min
max	Maximum value	max
approx_unique	Approximate distinct count (HyperLogLog)	unique_count
avg	Average value	mean
std	Standard deviation	std
q25	25th percentile	q25
q50	50th percentile (median)	q50
q75	75th percentile	q75
count	Non-null count	count
null_percentage	Percentage of null values	Derived null_count

The null count is derived from the null percentage and total row count:

null_count = int((null_pct / 100.0) * total_rows) if null_pct else 0

Fallback Schema

If SUMMARIZE fails (for example, on tables with unsupported column types or corrupt data), the system falls back to a basic schema query using information_schema.columns:

col_rows = con.execute(
    f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{table_name}'"
).fetchall()

The fallback provides column names and types but no statistics (null_count, unique_count, min, max, etc. are all set to zero or None). This ensures the schema endpoint always returns a valid response.

Column Type Classification

After DuckDB reports the raw data type, the classify_column() function in data/types.py refines the classification into one of six semantic categories:

ColumnType	Description	Example DuckDB Types
numeric	Quantitative values	INTEGER, BIGINT, FLOAT, DOUBLE, DECIMAL
categorical	Low-cardinality strings	VARCHAR with unique_count <= 50 or < 5% of rows
datetime	Temporal values	DATE, TIMESTAMP, TIME
boolean	True/false values	BOOLEAN
text	High-cardinality strings	VARCHAR with unique_count > 50 and > 5% of rows
geo_lat_lon	Geographic coordinates	Any type with column name matching lat/lon/latitude/longitude

Classification Logic

The classification proceeds in two steps:

1. Base type mapping: The DuckDB type string is matched against a prefix map. For example, VARCHAR maps to TEXT, INTEGER maps to NUMERIC, TIMESTAMP maps to DATETIME.

2. Refinement rules:

   • VARCHAR columns with 50 or fewer unique values, or where unique values represent less than 5% of total rows, are reclassified from TEXT to CATEGORICAL. This heuristic correctly identifies status fields, country codes, and other low-cardinality string columns.
   • VARCHAR or numeric columns whose name matches geographic identifiers (latitude, lat, longitude, lon, lng, case-insensitive) are reclassified as GEO_LAT_LON. This enables automatic map visualization detection.

if base == ColumnType.TEXT:
    if unique_count <= 50 or (total_rows > 0 and unique_count / total_rows < 0.05):
        return ColumnType.CATEGORICAL
    lower_name = name.lower()
    if lower_name in ("latitude", "lat", "longitude", "lon", "lng"):
        return ColumnType.GEO_LAT_LON
    return ColumnType.TEXT

Statistics Computation

DuckDB-Based Statistics (Primary)

All statistics are computed by DuckDB during the SUMMARIZE call. This approach has several advantages:

• Single-pass computation: DuckDB computes all statistics in one table scan, which is significantly faster than computing each statistic individually.
• Approximate algorithms: The approx_unique field uses HyperLogLog, providing fast cardinality estimation with low memory overhead.
• Native NULL handling: DuckDB's aggregate functions inherently exclude NULL values, ensuring correct statistical results.

Polars-Based Statistics (Supplementary)

The column_stats_polars() function provides an alternative statistics path using Polars' describe() method:

def column_stats_polars(df: pl.DataFrame, col_name: str) -> dict[str, Any]:
    col = df[col_name]
    stats = {
        "name": col_name,
        "dtype": str(col.dtype),
        "null_count": col.null_count(),
        "unique_count": col.n_unique(),
    }
    if col.dtype.is_numeric():
        desc = col.describe()
        stats.update({
            "mean": desc.filter(pl.col("statistic") == "mean")["value"][0],
            "std": desc.filter(pl.col("statistic") == "std")["value"][0],
            ...
        })
    return stats

This function is available for scenarios where the data resides in a Polars DataFrame rather than a DuckDB table, but it is not used in the current API flow.

Data Export

CSV Export

The export_to_csv() function uses DuckDB's COPY command to export a table to CSV:

def export_to_csv(con, table_name, output_path):
    con.execute(f'COPY "{table_name}" TO \'{output_path}\' (HEADER, DELIMITER \',\')')
    return output_path

The export always includes a header row and uses comma delimiters. The output file is written to the DATA_DIR directory with a unique filename prefix, and the file is served to the client as a FileResponse with text/csv media type.

Parquet Export

For checkpoint persistence, the export_to_parquet() function writes data in Parquet format:

def export_to_parquet(con, table_name, output_path):
    con.execute(f'COPY "{table_name}" TO \'{output_path}\' (FORMAT PARQUET)')
    return output_path

Parquet is preferred for checkpoints because of its columnar compression, type preservation, and significantly smaller file sizes compared to CSV. DuckDB's native Parquet writer preserves all type information, ensuring lossless round-tripping.

Polars Fallback Export

If the current DataFrameRef uses the Polars backend (which should not happen in normal operation because Polars results are materialized to DuckDB), the export falls back to Polars' CSV writer:

if ref.backend == "duckdb" and ref.name and ref.con:
    export_to_csv(ref.con, ref.name, str(output_path))
else:
    df = ref.to_polars_df()
    df.write_csv(str(output_path))