Spaces:
Sleeping
Sleeping
File size: 5,082 Bytes
c21180c 8ac8526 c21180c 8ac8526 c21180c 8ac8526 c21180c 8ac8526 c21180c 8ac8526 c21180c 8ac8526 c21180c 8ac8526 c21180c 8ac8526 c21180c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """Schema inference and column statistics."""
from __future__ import annotations
from typing import Any
import duckdb
import polars as pl
from visual_notebook.data.types import ColumnType, classify_column
def infer_schema_duckdb(con: duckdb.DuckDBPyConnection, table_name: str) -> list[dict[str, Any]]:
"""Infer schema and statistics using DuckDB SUMMARIZE."""
try:
rows = con.execute(f'SUMMARIZE "{table_name}"').fetchall()
# SUMMARIZE returns: column_name, column_type, min, max, approx_unique, avg, std, q25, q50, q75, count, null_percentage
columns = []
col_names = [d[0] for d in con.execute(f'SUMMARIZE "{table_name}"').description]
total_rows = con.execute(f'SELECT COUNT(*) FROM "{table_name}"').fetchone()[0]
for row in rows:
col = dict(zip(col_names, row))
col_name = col.get("column_name", col.get(col_names[0], ""))
col_type = str(col.get("column_type", ""))
null_pct = col.get("null_percentage", 0.0)
null_count = int((null_pct / 100.0) * total_rows) if null_pct else 0
unique_count = col.get("approx_unique", 0) or 0
classified = classify_column(col_type, col_name, null_count, unique_count, total_rows)
# Determine if numeric and compute median/iqr
is_numeric = classified in (ColumnType.NUMERIC, ColumnType.GEO_LAT_LON)
q50_val = col.get("q50")
q25_val = col.get("q25")
q75_val = col.get("q75")
median_val = None
iqr_val = None
if is_numeric and q50_val is not None:
median_val = float(q50_val)
if is_numeric and q25_val is not None and q75_val is not None:
iqr_val = float(q75_val) - float(q25_val)
columns.append({
"name": col_name,
"dtype": col_type,
"column_type": classified.value,
"null_count": null_count,
"unique_count": unique_count,
"min": col.get("min"),
"max": col.get("max"),
"mean": col.get("avg"),
"std": col.get("std"),
"q25": col.get("q25"),
"q50": col.get("q50"),
"q75": col.get("q75"),
"count": col.get("count", total_rows),
"has_numeric_stats": is_numeric,
"median": median_val,
"iqr": iqr_val,
})
return columns
except Exception:
# Fallback: basic column info
col_rows = con.execute(
f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{table_name}'"
).fetchall()
total_rows = con.execute(f'SELECT COUNT(*) FROM "{table_name}"').fetchone()[0]
columns = []
for name, dtype in col_rows:
classified = classify_column(dtype, name, 0, 0, total_rows)
is_numeric = classified in (ColumnType.NUMERIC, ColumnType.GEO_LAT_LON)
columns.append({
"name": name,
"dtype": dtype,
"column_type": classified.value,
"null_count": 0,
"unique_count": 0,
"min": None,
"max": None,
"mean": None,
"std": None,
"q25": None,
"q50": None,
"q75": None,
"count": total_rows,
"has_numeric_stats": is_numeric,
"median": None,
"iqr": None,
})
return columns
def column_stats_polars(df: pl.DataFrame, col_name: str) -> dict[str, Any]:
"""Compute detailed stats for a column using Polars."""
col = df[col_name]
classified = classify_column(str(col.dtype), col_name, col.null_count(), col.n_unique(), df.height)
is_numeric = classified in (ColumnType.NUMERIC, ColumnType.GEO_LAT_LON)
stats = {
"name": col_name,
"dtype": str(col.dtype),
"null_count": col.null_count(),
"unique_count": col.n_unique(),
"has_numeric_stats": is_numeric,
"median": None,
"iqr": None,
}
if col.dtype.is_numeric():
desc = col.describe()
q25 = desc.filter(pl.col("statistic") == "25%")["value"][0]
q50 = desc.filter(pl.col("statistic") == "50%")["value"][0]
q75 = desc.filter(pl.col("statistic") == "75%")["value"][0]
stats.update({
"mean": desc.filter(pl.col("statistic") == "mean")["value"][0],
"std": desc.filter(pl.col("statistic") == "std")["value"][0],
"min": desc.filter(pl.col("statistic") == "min")["value"][0],
"max": desc.filter(pl.col("statistic") == "max")["value"][0],
"q25": q25,
"q50": q50,
"q75": q75,
"median": float(q50) if q50 is not None else None,
"iqr": float(q75) - float(q25) if q25 is not None and q75 is not None else None,
})
return stats
|