search: index business entities + dataset quick-filters

Build dedicated de-duplicated entity indices (atc-customers ~39k,
atc-employees ~39k, atc-products) from Trino so users can search by
customer/employee name, id and email directly instead of scanning raw
transaction rows. Raise per-table sample caps for the big source tables
(sales_orders, employee_events, supplychain events, device_metrics) and
all iceberg/Hadoop tables for broader row-level coverage.

UI: add Datasets quick-filter chips (Customers, Employees, Products,
Sales orders, HR events, Supply chain, Telemetry, Lakehouse) and default
the Search tab to business data instead of an infra-looking empty view.

api/elasticsearch_api.py

@@ -30,7 +30,36 @@ INDEX_PREFIX = "atc-"
 # Trino catalogs to index (skip system + streaming kafka).
 INDEX_CATALOGS = ("postgres_sales", "mysql_hr", "mongodb_supplychain", "cassandra_telemetry", "iceberg")
 SKIP_SCHEMAS = {"information_schema", "sys", "performance_schema", "config", "local", "admin"}
-ROWS_PER_TABLE = int(os.getenv("ES_ROWS_PER_TABLE", "500"))
+ROWS_PER_TABLE = int(os.getenv("ES_ROWS_PER_TABLE", "1000"))
+# big source tables we want broadly searchable (row-level)
+ROWS_BUSINESS = int(os.getenv("ES_ROWS_BUSINESS", "15000"))
+BUSINESS_TABLES = {
+    ("postgres_sales", "public", "sales_orders"),
+    ("mysql_hr", "hr", "employee_events"),
+    ("mongodb_supplychain", "supplychain", "events"),
+    ("cassandra_telemetry", "telemetry", "device_metrics"),
+}
+# de-duplicated business entities so users can search people/customers by name + id
+ENTITY_QUERIES = [
+    (
+        "atc-customers", "customers", "customer",
+        'SELECT DISTINCT customer_id, customer_name, customer_email, region, sales_channel, currency '
+        'FROM "postgres_sales"."public"."sales_orders" WHERE customer_id IS NOT NULL LIMIT 50000',
+        "customer_id",
+    ),
+    (
+        "atc-employees", "employees", "employee",
+        'SELECT DISTINCT employee_id, employee_name, employee_email, employee_phone, department, role_name, region '
+        'FROM "mysql_hr"."hr"."employee_events" WHERE employee_id IS NOT NULL LIMIT 50000',
+        "employee_id",
+    ),
+    (
+        "atc-products", "products", "product",
+        'SELECT DISTINCT product_id, region, sales_channel FROM "postgres_sales"."public"."sales_orders" '
+        'WHERE product_id IS NOT NULL LIMIT 50000',
+        "product_id",
+    ),
+]
 
 router = APIRouter(prefix="/api/search", tags=["search"])
 
@@ -316,8 +345,8 @@ def _meta(catalog: str, schema: str, table: str, source: str) -> dict:
     }
 
 
-def _index_trino_table(sqlmod, client: httpx.Client, catalog: str, schema: str, table: str) -> int:
-    res = sqlmod._run_trino(f'SELECT * FROM "{catalog}"."{schema}"."{table}" LIMIT {ROWS_PER_TABLE}', ROWS_PER_TABLE)
+def _index_trino_table(sqlmod, client: httpx.Client, catalog: str, schema: str, table: str, limit: int = ROWS_PER_TABLE) -> int:
+    res = sqlmod._run_trino(f'SELECT * FROM "{catalog}"."{schema}"."{table}" LIMIT {limit}', limit)
     if not res.get("ok"):
         raise RuntimeError(res.get("error", "query failed")[:200])
     cols = res.get("columns", [])
@@ -334,6 +363,37 @@ def _index_trino_table(sqlmod, client: httpx.Client, catalog: str, schema: str,
     return written
 
 
+def _index_entities(sqlmod, client: httpx.Client) -> int:
+    """Build de-duplicated entity indices (customers, employees, products) so users
+    can search by name / id directly instead of scanning raw transaction rows."""
+    written = 0
+    for index, catalog_label, schema_label, sql, id_field in ENTITY_QUERIES:
+        _reindex_state["current"] = f"entities · {catalog_label}"
+        try:
+            res = sqlmod._run_trino(sql, 50000)
+            if not res.get("ok"):
+                _reindex_state["errors"].append(f"{index}: {res.get('error', 'failed')[:120]}")
+                continue
+            cols = res.get("columns", [])
+            rows = res.get("rows", [])
+            docs = []
+            for row in rows:
+                doc = {cols[i]: row[i] for i in range(min(len(cols), len(row)))}
+                ident = doc.get(id_field)
+                doc["meta"] = _meta(catalog_label, schema_label, id_field, catalog_label)
+                docs.append((f"{schema_label}:{ident}", doc))
+            n = 0
+            for i in range(0, len(docs), 500):
+                n += _bulk(client, index, docs[i:i + 500])
+            _reindex_state["indices"][index] = n
+            _reindex_state["total_docs"] += n
+            _reindex_state["log"].append(f"{index} → {n} entities")
+            written += n
+        except Exception as exc:  # noqa: BLE001
+            _reindex_state["errors"].append(f"{index}: {str(exc)[:120]}")
+    return written
+
+
 def _index_neo4j(sqlmod, client: httpx.Client) -> int:
     written = 0
     try:
@@ -396,6 +456,9 @@ def _reindex_worker():
             _reindex_state["current"] = "catalog inventory"
             _reindex_state["total_docs"] += _index_catalog(client)
 
+            # business entities (customers / employees / products) — the headline searchables
+            _index_entities(sqlmod, client)
+
             # Trino-federated source data
             cat_res = sqlmod._run_trino("SHOW CATALOGS", 100)
             catalogs = [r[0] for r in cat_res.get("rows", [])] if cat_res.get("ok") else list(INDEX_CATALOGS)
@@ -420,7 +483,8 @@ def _reindex_worker():
                     for table in tables:
                         _reindex_state["current"] = f"{catalog}.{schema}.{table}"
                         try:
-                            n = _index_trino_table(sqlmod, client, catalog, schema, table)
+                            cap = ROWS_BUSINESS if (catalog, schema, table) in BUSINESS_TABLES or catalog == "iceberg" else ROWS_PER_TABLE
+                            n = _index_trino_table(sqlmod, client, catalog, schema, table, cap)
                             idx = _sanitize_index(catalog, schema, table)
                             _reindex_state["indices"][idx] = n
                             _reindex_state["total_docs"] += n