Blame for 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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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`:

```python
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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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))
```