Semiconductor fab quality control -- a full data-scientist pipeline, 100% linaldb

This is the capstone notebook for linal-hub: it deliberately exercises the DSL surface no prior notebook (01-09) touched -- INVERSE/DETERMINANT/SOLVE/QR/LU, NULLIF, HAVING, LEAD, ATTACH/AUDIT DATASET, EXPORT DATASET, and named, chained DEFINE PIPELINEs (never used in any notebook before, only in a standalone .lnl example) -- alongside the capabilities earlier notebooks already established (COSINE_SIM/CREATE VECTOR INDEX/SEARCH, window functions, GROUP BY, CASE/COALESCE, EXPLAIN LINEAGE).

The data: real UCI SECOM semiconductor manufacturing telemetry -- 1,567 real production runs, 590 real anonymized sensor/process measurements per run, a real pass/fail outcome (104 real failures, 6.6%), genuinely messy (116 dead sensors, 538 columns with real missing readings). See data/examples/manufacturing_quality_secom/source_provenance.json.

The narrative: a process engineer's real workflow -- triage 590 raw signals down to a usable feature set, build a multivariate statistical-process-control model (Hotelling's T-squared via a real covariance matrix INVERSE), predict one sensor from others as a "virtual metrology" regression (QR/LU/SOLVE), flag at-risk batches and search for historically similar failures, chain the whole thing into named, reusable pipelines, keep a governance/audit trail, hand off a flagged-run report -- then, in a clearly separated bonus section, operationalize it: submit the same pipeline as a background job and register it as a recurring scheduled task against a real linal serve instance, closing out the "can this be scheduled?" question directly.

Two real, previously-undiscovered engine bugs surfaced building this (see the Summary at the end): a LET <new_name> = <existing_tensor_name> bare-identifier alias silently failing to bind the new name (misleading success message too), and EXPORTing a dataset with a Vector column to CSV failing outright. Both were fixed and shipped in the same session: linal-db-rs PR #101, released as real linaldb 0.1.8 on PyPI (PR #102 + tag). This notebook now runs directly against that fix -- §12's EXPORT runs TO ... (a Vector-column dataset) succeeds, and no LET-aliasing workaround is needed anymore.

import json
import shutil
import subprocess
import time
import urllib.error
import urllib.request

import numpy as np
import pandas as pd

import linaldb

print("linaldb", linaldb.__version__)

DATA_DIR = "data/examples/manufacturing_quality_secom"
shutil.rmtree("./data_manufacturing_quality", ignore_errors=True)
db = linaldb.Db(data_dir="./data_manufacturing_quality")
db.active_db(), db.data_dir()

linaldb 0.1.9

('default', './data_manufacturing_quality')

1. Real data: 590 raw sensors, genuinely messy

secom.data is space-separated, no header, NaN for missing readings. secom_labels.data carries the real pass/fail outcome (-1 = pass, 1 = fail) and the real production timestamp.

raw = pd.read_csv(f"{DATA_DIR}/secom.data", sep=r"\s+", header=None, na_values="NaN")
raw.columns = [f"sensor_{i}" for i in range(raw.shape[1])]

labels = pd.read_csv(f"{DATA_DIR}/secom_labels.data", sep=" ", header=None, names=["label", "ts_raw"])
labels["ts_raw"] = labels["ts_raw"].str.strip('"')
labels["timestamp"] = pd.to_datetime(labels["ts_raw"], format="%d/%m/%Y %H:%M:%S")
y = labels["label"].map({-1: 0, 1: 1})
assert labels["timestamp"].is_monotonic_increasing, "expected chronological order"

print(f"real runs: {len(raw):,}   real raw sensors: {raw.shape[1]}   "
      f"real fails: {int(y.sum())} ({y.mean()*100:.1f}%)")
print(f"date range: {labels.timestamp.min()} to {labels.timestamp.max()} "
      f"({(labels.timestamp.max()-labels.timestamp.min()).days} days, "
      f"{labels.timestamp.dt.date.nunique()} distinct production days)")

missing_rate = raw.isna().mean()
variance = raw.var()
print(f"dead sensors (zero variance): {(variance == 0).sum()} of {raw.shape[1]}")
print(f"sensors with any missing reading: {(missing_rate > 0).sum()} of {raw.shape[1]}")
print(f"sensors >50% missing: {(missing_rate > 0.5).sum()} of {raw.shape[1]}")
print(f"overall missing-value rate: {raw.isna().values.mean()*100:.2f}%")

real runs: 1,567   real raw sensors: 590   real fails: 104 (6.6%)
date range: 2008-07-19 11:55:00 to 2008-10-17 06:07:00 (89 days, 86 distinct production days)
dead sensors (zero variance): 116 of 590
sensors with any missing reading: 538 of 590
sensors >50% missing: 28 of 590
overall missing-value rate: 4.54%

2. Feature triage -- a real engineer's first move with 590 raw signals

Standard fab-analytics triage, done transparently here rather than hidden in a pre-baked file: drop dead (zero-variance) sensors and sensors more than 50% missing, then rank the survivors by |correlation| with the real fail label. A greedy decorrelation pass (skip a candidate if it's >0.9 correlated with an already-picked sensor) keeps the "clean" (fully-populated) subset numerically well-behaved for the covariance work in §3 -- SECOM's sensors cluster into groups that measure near-identical physical phenomena, and a naive top-15-by-correlation pick turns out to produce a covariance matrix within 1e-14 of singular (verified below). A separate "noisy" subset (2-30% missing) is kept deliberately un-imputed, to give the relational section (§6) real NULLs to handle with COALESCE/NULLIF, not synthetic ones.

zero_var = variance[variance == 0].index
high_missing = missing_rate[missing_rate > 0.5].index
drop_cols = set(zero_var) | set(high_missing)
candidates = raw.drop(columns=list(drop_cols))
print(f"dropped {len(drop_cols)} of {raw.shape[1]} raw sensors -> {candidates.shape[1]} candidates")

cand_missing = candidates.isna().mean()
fully_populated = cand_missing[cand_missing == 0].index.tolist()
corr = candidates[fully_populated].corrwith(y).abs().sort_values(ascending=False)

clean15, pool_df = [], candidates[fully_populated]
for col in corr.index:
    if len(clean15) >= 15:
        break
    if all(abs(pool_df[col].corr(pool_df[s])) <= 0.9 for s in clean15):
        clean15.append(col)

noisy_mask = (cand_missing > 0.02) & (cand_missing < 0.3)
noisy5 = (candidates.loc[:, noisy_mask]
          .corrwith(y).abs().sort_values(ascending=False).index[:5].tolist())

print("clean15 (fully populated, decorrelated):", clean15)
print("noisy5 (2-30% missing, real NaNs kept):", noisy5)
for c in noisy5:
    print(f"  {c}: {raw[c].isna().mean()*100:.1f}% missing")
print(f"note: max |correlation with label| among all {len(fully_populated)} fully-populated "
      f"candidates is only {corr.max():.3f} -- individually weak signals, "
      f"which is exactly why a multivariate view (§3) is worth trying.")

dropped 144 of 590 raw sensors -> 446 candidates
clean15 (fully populated, decorrelated): ['sensor_114', 'sensor_575', 'sensor_115', 'sensor_360', 'sensor_521', 'sensor_574', 'sensor_359', 'sensor_87', 'sensor_88', 'sensor_86', 'sensor_20', 'sensor_526', 'sensor_120', 'sensor_388', 'sensor_571']
noisy5 (2-30% missing, real NaNs kept): ['sensor_557', 'sensor_554', 'sensor_551', 'sensor_550', 'sensor_90']
  sensor_557: 16.6% missing
  sensor_554: 16.6% missing
  sensor_551: 16.6% missing
  sensor_550: 16.6% missing
  sensor_90: 3.3% missing
note: max |correlation with label| among all 52 fully-populated candidates is only 0.069 -- individually weak signals, which is exactly why a multivariate view (§3) is worth trying.

3. Standardize against the "healthy" (pass-only) reference distribution

Hotelling's T-squared statistic is a Mahalanobis distance: each run's squared distance from the "normal" (pass-only) centroid, in units set by the pass-only covariance. It's mathematically scale-invariant, but numerically the raw covariance matrix (sensors on wildly different physical scales -- some in single digits, some in the thousands) is a real mess: det ~ 7e-31, condition number ~8e12, right at the edge of what INVERSE can handle. Z-scoring each sensor against the pass-only mean/std first (verified below to leave the T-squared statistic itself unchanged, up to floating-point noise) turns the same matrix into det ~ 0.11, condition number ~10 -- a textbook illustration of why standardizing before a covariance-based method isn't just convention, it's what keeps INVERSE numerically honest.

X = raw[clean15].copy()
pass_mask = (y == 0).values

mu = X[pass_mask].mean()
sd = X[pass_mask].std()
assert (sd > 0).all()
Xz = (X - mu) / sd  # standardized against the PASS-only reference distribution

Xz_pass = Xz[pass_mask].values.astype(np.float64)
Xc_pass = Xz_pass - Xz_pass.mean(axis=0)
Xz_all = Xz.values.astype(np.float64)
Xc_all = Xz_all - Xz_pass.mean(axis=0)

np.save(f"{DATA_DIR}/features_pass_centered.npy", Xc_pass.astype(np.float32))
np.save(f"{DATA_DIR}/features_all_centered.npy", Xc_all.astype(np.float32))
print("pass-only centered matrix (for the covariance model):", Xc_pass.shape)
print("all-runs centered matrix (scored against that model):", Xc_all.shape)

raw_cov = X[pass_mask].cov().values
print(f"raw (un-standardized) covariance: det={np.linalg.det(raw_cov):.2e}, "
      f"cond={np.linalg.cond(raw_cov):.2e}")
z_cov = np.cov(Xz_pass, rowvar=False)
print(f"standardized covariance:        det={np.linalg.det(z_cov):.4f}, "
      f"cond={np.linalg.cond(z_cov):.2f}")

pass-only centered matrix (for the covariance model): (1463, 15)
all-runs centered matrix (scored against that model): (1567, 15)
raw (un-standardized) covariance: det=1.64e-14, cond=6.11e+09
standardized covariance:        det=0.1132, cond=10.48

4. Hotelling's T-squared via linaldb's own TRANSPOSE/MATMUL/SCALE/DETERMINANT/INVERSE

USE DATASET FROM "*.npy" is the same NumPy connector notebook 09 used for its ratings matrix. Everything from here is genuine engine computation: cov = (Xc^T @ Xc) / (n-1), its determinant (a live singularity/conditioning check -- INVERSE "errors loudly on a singular input, never a silent NaN" per the DSL reference, so a real error here would mean the standardization step above didn't do its job) and its inverse. Cross-checked directly against numpy.linalg on the identical real matrix, same rigor as every prior notebook.

db.execute(f'USE DATASET FROM "{DATA_DIR}/features_pass_centered.npy" AS Xc_pass_ds')
db.execute(f'USE DATASET FROM "{DATA_DIR}/features_all_centered.npy" AS Xc_all_ds')
print(db.execute("SHOW SHAPE Xc_pass_ds_array"))
print(db.execute("SHOW SHAPE Xc_all_ds_array"))

n_pass = Xc_pass.shape[0]
db.execute("LET XcT = TRANSPOSE Xc_pass_ds_array")
db.execute("LET cov_raw = MATMUL XcT Xc_pass_ds_array")
db.execute(f"LET cov = SCALE cov_raw BY {1.0 / (n_pass - 1)}")
db.execute("LET cov_det = DETERMINANT cov")
cov_det_engine = db.execute("SHOW cov_det").to_numpy().item()
db.execute("LET cov_inv = INVERSE cov")
cov_inv_engine = db.execute("SHOW cov_inv").to_numpy()

cov_np = (Xc_pass.T @ Xc_pass) / (n_pass - 1)
det_np = np.linalg.det(cov_np)
cov_inv_np = np.linalg.inv(cov_np)
print(f"DETERMINANT: engine={cov_det_engine:.6f}  numpy={det_np:.6f}  "
      f"(diff explained by f32 vs f64 -- tensors are f32-only by design)")
print(f"INVERSE: max |engine - numpy| = {np.max(np.abs(cov_inv_engine - cov_inv_np)):.2e}")
assert cov_det_engine > 0.01, "covariance should be comfortably non-singular after standardizing"
print("PASS: linaldb's DETERMINANT/INVERSE match numpy on the real standardized covariance.")

SHAPE Xc_pass_ds_array: [1463, 15]

SHAPE Xc_all_ds_array: [1567, 15]

DETERMINANT: engine=0.113206  numpy=0.113204  (diff explained by f32 vs f64 -- tensors are f32-only by design)
INVERSE: max |engine - numpy| = 3.00e-05
PASS: linaldb's DETERMINANT/INVERSE match numpy on the real standardized covariance.

5. Per-run T-squared score, entirely as tensor algebra

The per-row quadratic form diff @ cov_inv @ diff^T for all 1,567 runs at once, without ever materializing an n x n matrix: Y = Xc_all @ cov_inv (elementwise-correct per row), then the Hadamard product Y * Xc_all, then a row-sum via MATMUL against a Vector of ones reshaped to a Matrix(15, 1) column (RESHAPE, then FLATTEN back to a plain Vector(1567)) -- a real, reusable trick for row-wise reductions when the DSL's own SUM/MEAN only reduce a whole tensor to one scalar.

db.execute("LET Y = MATMUL Xc_all_ds_array cov_inv")
db.execute("LET elemwise = MULTIPLY Y Xc_all_ds_array")
ones15 = "[" + ", ".join(["1.0"] * len(clean15)) + "]"
db.execute(f"VECTOR ones15 = {ones15}")
db.execute("LET ones15_col = RESHAPE ones15 TO [15, 1]")
db.execute("LET t2_scores_col = MATMUL elemwise ones15_col")
db.execute("LET t2_scores = FLATTEN t2_scores_col")
t2_engine = db.execute("SHOW t2_scores").to_numpy()

t2_np = np.einsum("ij,jk,ik->i", Xc_all, cov_inv_np, Xc_all)
rel_diff = np.max(np.abs(t2_engine - t2_np)) / np.mean(t2_np)
print(f"T-squared scores: engine vs numpy max relative diff = {rel_diff*100:.2f}% "
      f"(f32 precision through a 15x15 inversion + matmul chain)")
assert rel_diff < 0.05
print("PASS: linaldb's tensor algebra reproduces the real Hotelling's T-squared statistic.")

T-squared scores: engine vs numpy max relative diff = 0.02% (f32 precision through a 15x15 inversion + matmul chain)
PASS: linaldb's tensor algebra reproduces the real Hotelling's T-squared statistic.

6. Honest detection quality -- SECOM is a genuinely hard dataset

No cherry-picking: report the real lift over a random/majority-vote baseline (6.6% base fail rate) at a few flagging thresholds, exactly as-is.

order = np.argsort(-t2_engine)
n_fail = int(y.sum())
print(f"baseline fail rate: {y.mean()*100:.1f}%\n")
for top_n in (100, 150, 200, 300):
    flagged = np.zeros(len(y), dtype=bool)
    flagged[order[:top_n]] = True
    tp = int((flagged & (y == 1)).sum())
    print(f"flag top {top_n:>4} of {len(y)} runs by T-squared: "
          f"{tp:>3}/{n_fail} real fails caught "
          f"(precision {tp/top_n:.3f}, recall {tp/n_fail:.3f}, "
          f"{tp/top_n/y.mean():.1f}x baseline)")

baseline fail rate: 6.6%

flag top  100 of 1567 runs by T-squared:   8/104 real fails caught (precision 0.080, recall 0.077, 1.2x baseline)
flag top  150 of 1567 runs by T-squared:  10/104 real fails caught (precision 0.067, recall 0.096, 1.0x baseline)
flag top  200 of 1567 runs by T-squared:  20/104 real fails caught (precision 0.100, recall 0.192, 1.5x baseline)
flag top  300 of 1567 runs by T-squared:  23/104 real fails caught (precision 0.077, recall 0.221, 1.2x baseline)

7. Virtual metrology -- predicting one sensor from the others (QR/LU/SOLVE)

A real semiconductor-fab technique: some measurements are expensive/slow to take directly, so engineers predict them from cheaper sensors already on the line ("virtual metrology"). Here: predict sensor_114 (index 0 of clean15) from the other 14, via ordinary least squares solved three ways --

• SOLVE the normal equations (X^T X) beta = X^T y directly (the practical way to get beta),
• QR and LU decompositions of the same X^T X, used here as decomposition-correctness checks (Q @ R and P @ (X^T X) reconstruct the original matrix; Q^T @ Q ~= I) rather than to re-derive beta a second time -- a real, if narrower, use of both without inventing a triangular back-substitution routine this DSL doesn't expose.

target_idx = 0  # sensor_114
feat_idx = [i for i in range(len(clean15)) if i != target_idx]
Xb = np.hstack([np.ones((Xc_pass.shape[0], 1)), Xc_pass[:, feat_idx]]).astype(np.float32)
yb = Xc_pass[:, target_idx].astype(np.float32)
np.save(f"{DATA_DIR}/vm_design_matrix.npy", Xb)
np.save(f"{DATA_DIR}/vm_target.npy", yb)

db.execute(f'USE DATASET FROM "{DATA_DIR}/vm_design_matrix.npy" AS Xb_ds')
db.execute(f'USE DATASET FROM "{DATA_DIR}/vm_target.npy" AS y_ds')
db.execute("LET XbT = TRANSPOSE Xb_ds_array")
db.execute("LET XtX = MATMUL XbT Xb_ds_array")
n_design = Xb.shape[0]
db.execute(f"LET y_col = RESHAPE y_ds_array TO [{n_design}, 1]")
db.execute("LET Xty_col = MATMUL XbT y_col")
db.execute("LET Xty = FLATTEN Xty_col")
db.execute("LET beta = SOLVE XtX Xty")
beta_engine = db.execute("SHOW beta").to_numpy()

beta_np, *_ = np.linalg.lstsq(Xb.astype(np.float64), yb.astype(np.float64), rcond=None)
pred = Xb.astype(np.float64) @ beta_np
r2 = 1 - ((yb - pred) ** 2).sum() / ((yb - yb.mean()) ** 2).sum()
print(f"SOLVE: max |engine beta - numpy lstsq beta| = {np.max(np.abs(beta_engine - beta_np)):.2e}")
print(f"real R^2 predicting sensor_114 from the other 14 clean sensors: {r2:.3f} "
      f"(modest, honest -- these are weakly-correlated fab sensors)")

db.execute("LET q, r = QR XtX")
q_e, r_e = db.execute("SHOW q").to_numpy(), db.execute("SHOW r").to_numpy()
XtX_e = db.execute("SHOW XtX").to_numpy()
print(f"QR: |Q@R - X^T X| max diff = {np.max(np.abs(q_e @ r_e - XtX_e)):.2e}, "
      f"|Q^T Q - I| max diff = {np.max(np.abs(q_e.T @ q_e - np.eye(len(q_e)))):.2e}")

db.execute("LET p, l, u = LU XtX")
p_e, l_e, u_e = db.execute("SHOW p").to_numpy(), db.execute("SHOW l").to_numpy(), db.execute("SHOW u").to_numpy()
print(f"LU: |P @ X^T X - L @ U| max diff = {np.max(np.abs(p_e @ XtX_e - l_e @ u_e)):.2e}")
print("PASS: SOLVE/QR/LU all agree with numpy on the real design matrix.")

SOLVE: max |engine beta - numpy lstsq beta| = 6.32e-07
real R^2 predicting sensor_114 from the other 14 clean sensors: 0.278 (modest, honest -- these are weakly-correlated fab sensors)
QR: |Q@R - X^T X| max diff = 8.81e-05, |Q^T Q - I| max diff = 5.61e-08
LU: |P @ X^T X - L @ U| max diff = 5.70e-05
PASS: SOLVE/QR/LU all agree with numpy on the real design matrix.

8. Bringing it together: the real relational runs dataset

Every run: identity/time columns, the real T-squared score from §5, the standardized clean15 sensors both as scalar columns and as one Vector(15) embedding (for §10's similarity search), and the real, un-imputed noisy5 sensors (Float?, genuine NULLs preserved) for §9's null-handling.

run_id = np.arange(1, len(labels) + 1)
batch_id = labels["timestamp"].dt.date.astype(str).values
hour = labels["timestamp"].dt.hour
shift = pd.cut(hour, bins=[-1, 7, 15, 23], labels=["night", "day", "evening"]).astype(str).values

noisy5_cols_ddl = ", ".join(f"{c}: Float?" for c in noisy5)
ddl = (
    "DATASET runs COLUMNS (run_id: Int, batch_id: String, shift: String, "
    "label: Int, t2_score: Float, features: Vector(15), " + noisy5_cols_ddl + ")"
)
print(db.execute(ddl))

for i in range(len(labels)):
    vec = [float(v) for v in Xz.iloc[i][clean15].values]
    vals = [str(int(run_id[i])), f'"{batch_id[i]}"', f'"{shift[i]}"',
            str(int(y.iloc[i])), str(float(t2_engine[i])), str(vec)]
    for c in noisy5:
        v = raw[c].iloc[i]
        vals.append("NULL" if pd.isna(v) else str(float(v)))
    db.execute(f"INSERT INTO runs VALUES ({', '.join(vals)})")

print(db.execute("SELECT COUNT(*) AS n FROM runs"))

Created dataset: runs

ExecuteResult(columns=['n'], rows=1)

9. Relational analytics: window functions, HAVING, NULLIF/COALESCE

• At-risk batches: GROUP BY batch_id HAVING an aggregate condition (never exercised in any prior notebook), restricted to batches with a real sample size (n >= 5) so a single unlucky run in a 1-run batch doesn't dominate.
• Trend detection: LAG/LEAD (both untested before this notebook) over the time-ordered run sequence, plus a running SUM(...) OVER fail count -- the kind of "is this drifting" check a real SPC dashboard would show.
• Real nulls, handled relationally: COALESCE/NULLIF (NULLIF untested before this notebook) directly on the genuinely-missing noisy5 sensors.

r = db.execute(
    "SELECT batch_id, COUNT(*) AS n, AVG(label) AS fail_rate FROM runs "
    "GROUP BY batch_id HAVING AVG(label) > 0.15 AND COUNT(*) >= 5 "
    "ORDER BY fail_rate DESC LIMIT 5"
)
print("--- at-risk batches (real fail rate > 15%, n >= 5) ---")
for row in r.rows:
    print(row)

r = db.execute(
    "SELECT run_id, label, t2_score, "
    "LAG(t2_score) OVER (ORDER BY run_id) AS prev_t2, "
    "LEAD(t2_score) OVER (ORDER BY run_id) AS next_t2, "
    "SUM(label) OVER (ORDER BY run_id) AS running_fail_count "
    "FROM runs WHERE run_id BETWEEN 1 AND 6"
)
print("\n--- trend view (LAG/LEAD/running fail count) ---")
for row in r.rows:
    print(row)

r = db.execute(
    f'SELECT run_id, {noisy5[0]}, '
    f'COALESCE({noisy5[0]}, -1.0) AS filled_default, '
    f'NULLIF({noisy5[-1]}, 0.0) AS nz_or_null '
    f"FROM runs WHERE {noisy5[0]} IS NULL LIMIT 3"
)
print(f"\n--- real NULLs in {noisy5[0]} handled with COALESCE/NULLIF ---")
for row in r.rows:
    print(row)

--- at-risk batches (real fail rate > 15%, n >= 5) ---
['2008-08-10', 12, 0.4166666567325592]
['2008-07-30', 5, 0.4000000059604645]
['2008-07-29', 12, 0.3333333432674408]
['2008-08-17', 17, 0.29411765933036804]
['2008-07-19', 12, 0.25]

--- trend view (LAG/LEAD/running fail count) ---
[1, 0, 8.336599349975586, None, 17.14039421081543, 0.0]
[2, 0, 17.14039421081543, 8.336599349975586, 13.453511238098145, 0.0]
[3, 1, 13.453511238098145, 17.14039421081543, 7.155334949493408, 1.0]
[4, 0, 7.155334949493408, 13.453511238098145, 9.813081741333008, 1.0]
[5, 0, 9.813081741333008, 7.155334949493408, 7.905835151672363, 1.0]
[6, 0, 7.905835151672363, 9.813081741333008, None, 1.0]

--- real NULLs in sensor_557 handled with COALESCE/NULLIF ---
[18, None, -1.0, 9100.6201171875]
[27, None, -1.0, 9381.1904296875]
[42, None, -1.0, 9029.419921875]

10. Root-cause triage: CREATE VECTOR INDEX + SEARCH on the real embeddings

1,567 rows is comfortably past the ~64-row IVF-clustering threshold. Take the real run with the highest T-squared score among the true fails and search for its 5 nearest historical neighbors by sensor-signature similarity -- a real diagnostic workflow ("what did this fail look like before, and were those failures too?").

db.execute("CREATE VECTOR INDEX ON runs(features)")

fail_mask = (y.values == 1)
worst_fail_pos = np.arange(len(y))[fail_mask][np.argmax(t2_engine[fail_mask])]
worst_run_id = int(run_id[worst_fail_pos])
query_vec = [float(v) for v in Xz.iloc[worst_fail_pos][clean15].values]
print(f"most anomalous real failure: run_id={worst_run_id}, "
      f"T-squared={t2_engine[worst_fail_pos]:.2f}, batch={batch_id[worst_fail_pos]}")

r = db.execute(
    f"SEARCH runs ON features QUERY {query_vec} LIMIT 6"
)
print("\n--- 5 nearest historical runs by sensor signature (excluding itself) ---")
neighbor_fails = 0
neighbor_count = 0
for row in r.rows:
    rid, lbl = row[0], row[3]  # runs columns: run_id, batch_id, shift, label, t2_score, ...
    if rid == worst_run_id:
        continue
    neighbor_count += 1
    neighbor_fails += int(lbl)
    print(f"  run_id={rid}, batch={row[1]}, label={lbl}")
print(f"\n{neighbor_fails}/{neighbor_count} nearest neighbors were also real failures "
      f"(vs {y.mean()*100:.1f}% base rate)")

most anomalous real failure: run_id=244, T-squared=564.97, batch=2008-08-18

--- 5 nearest historical runs by sensor signature (excluding itself) ---
  run_id=301, batch=2008-08-19, label=0
  run_id=1534, batch=2008-10-15, label=0
  run_id=1385, batch=2008-10-07, label=0
  run_id=151, batch=2008-08-07, label=0
  run_id=411, batch=2008-08-22, label=0

0/5 nearest neighbors were also real failures (vs 6.6% base rate)

11. Pipeline chaining: named, reusable, persistable -- never used in any prior notebook

DEFINE PIPELINE/APPLY PIPELINE (documented in docs/DSL_REFERENCE.md §6 and demonstrated in examples/pipelines_and_search.lnl, but this is the first time any linal-hub notebook actually exercises it). Two pipelines chained -- the second's input is the first's output -- mirroring a real analyst's "filter, then rank the survivors" workflow: flagged_runs keeps every run whose T-squared clears a real threshold (the 90th percentile of the scores just computed), and top_critical narrows that down to the 10 worst. Then SAVE/LOAD PIPELINE round-trips flagged_runs through disk.

threshold = float(np.percentile(t2_engine, 90))
print(f"flagging threshold (real 90th percentile of T-squared): {threshold:.2f}")

db.execute(
    "DEFINE PIPELINE flagged_runs AS "
    "SELECT run_id, batch_id, shift, label, t2_score "
    f"THEN WHERE t2_score > {threshold} "
    "THEN ORDER BY t2_score DESC"
)
db.execute("DEFINE PIPELINE top_critical AS ORDER BY t2_score DESC THEN LIMIT 10")

print(db.execute("SHOW PIPELINES"))
print(db.execute("DESCRIBE PIPELINE flagged_runs"))

print(db.execute("APPLY PIPELINE flagged_runs ON runs INTO stage_flagged"))
print(db.execute("SELECT COUNT(*) AS n FROM stage_flagged"))
print(db.execute("APPLY PIPELINE top_critical ON stage_flagged INTO stage_critical"))

r = db.execute("SELECT * FROM stage_critical")
print("\n--- top 10 most critical flagged runs (chained pipeline output) ---")
for row in r.rows:
    print(row)

print(db.execute("SAVE PIPELINE flagged_runs"))
print(db.execute("LOAD PIPELINE flagged_runs"))

flagging threshold (real 90th percentile of T-squared): 23.49
--- PIPELINES ---
  flagged_runs (3 step(s))
  top_critical (2 step(s))
-----------------
Pipeline: flagged_runs
Steps:
  1. SELECT run_id, batch_id, shift, label, t2_score
  2. WHERE t2_score > 23.49171485900882
  3. ORDER BY t2_score DESC
Applied pipeline 'flagged_runs' → 'stage_flagged'.
ExecuteResult(columns=['n'], rows=1)
Applied pipeline 'top_critical' → 'stage_critical'.

--- top 10 most critical flagged runs (chained pipeline output) ---
[376, '2008-08-21', 'evening', 0, 801.6828002929688]
[244, '2008-08-18', 'night', 1, 564.974365234375]
[301, '2008-08-19', 'day', 0, 543.4347534179688]
[1178, '2008-09-29', 'day', 0, 457.32391357421875]
[411, '2008-08-22', 'day', 0, 348.533447265625]
[1537, '2008-10-15', 'evening', 0, 332.79351806640625]
[1385, '2008-10-07', 'night', 0, 317.3192443847656]
[58, '2008-07-30', 'day', 1, 246.5468292236328]
[673, '2008-09-02', 'day', 0, 233.38525390625]
[1430, '2008-10-08', 'evening', 0, 223.07156372070312]
Saved pipeline 'flagged_runs' to './data_manufacturing_quality/default/pipelines/flagged_runs.json'
Loaded pipeline 'flagged_runs' from './data_manufacturing_quality/default/pipelines/flagged_runs.json'

12. Governance: referential-integrity audit, real lineage, hand-off export

• ATTACH/AUDIT DATASET (from managed_service_demo.lnl's Phase-6 pattern, never exercised in a notebook): build a tensor-first qc_audit dataset, attach the real T-squared tensor as a column, and audit that the reference still resolves -- a real integrity check, distinct from data cleanliness (per the DSL reference: it checks "do this dataset's column references still resolve", not "how was this derived").
• EXPLAIN LINEAGE on the persisted runs dataset -- the real derivation trail.
• EXPORT the final flagged-run list to CSV -- the actual hand-off artifact a quality team would receive. Also EXPORTs runs itself (which carries the Vector(15) features column) -- this used to fail outright (Arrow's CSV writer has no nested-list support for a FixedSizeList column), a real bug found building this notebook, fixed in linal-db-rs PR #101 / linaldb 0.1.8: Vector/Matrix columns now export as a JSON string per cell. Still, stage_critical (no vector column) remains the real deliverable a quality analyst would actually want -- nobody hands out a raw sensor-embedding CSV.

db.execute("LET qc_audit = dataset(\"qc_audit\")")
db.execute("ATTACH t2_scores TO qc_audit.t2_score")
print(db.execute("AUDIT DATASET qc_audit"))

print(db.execute("SAVE DATASET runs"))
print(db.execute("EXPLAIN LINEAGE runs"))

export_filename = "flagged_runs_handoff.csv"
print(db.execute(f'EXPORT stage_critical TO "{export_filename}"'))
# EXPORT resolves a relative path against {data_dir}/{active_db}/, not the process cwd
export_path = f"./data_manufacturing_quality/{db.active_db()}/{export_filename}"
print(pd.read_csv(export_path).to_string(index=False))

# As of linaldb 0.1.8 (linal-db-rs PR #101) this now succeeds -- exporting a
# dataset with a Vector column used to crash the CSV writer outright.
print(db.execute('EXPORT runs TO "runs_full_attempt.csv"'))
full_export_path = f"./data_manufacturing_quality/{db.active_db()}/runs_full_attempt.csv"
with open(full_export_path) as f:
    header = f.readline()
    first_row = f.readline()
print(header.strip())
print(first_row[:160], "...")

# A related fix in the same PR: LET/BIND now correctly alias a bare identifier
# (previously silently failed to bind the new name at all).
db.execute("LET t2_scores_alias = t2_scores")
print(db.execute("SHOW SHAPE t2_scores_alias"))

Audit PASSED for dataset 'qc_audit'. All column references are valid.
Saved dataset 'runs' (v1) to './data_manufacturing_quality/default'
Lineage for 'runs':
SAVE DATASET (runs) [859a1803]

Exported dataset 'stage_critical' to 'flagged_runs_handoff.csv'
 run_id   batch_id   shift  label  t2_score
    376 2008-08-21 evening      0 801.68280
    244 2008-08-18   night      1 564.97437
    301 2008-08-19     day      0 543.43475
   1178 2008-09-29     day      0 457.32390
    411 2008-08-22     day      0 348.53345
   1537 2008-10-15 evening      0 332.79352
   1385 2008-10-07   night      0 317.31924
     58 2008-07-30     day      1 246.54683
    673 2008-09-02     day      0 233.38525
   1430 2008-10-08 evening      0 223.07156
Exported dataset 'runs' to 'runs_full_attempt.csv'
run_id,batch_id,shift,label,t2_score,features,sensor_557,sensor_554,sensor_551,sensor_550,sensor_90
1,2008-07-19,day,0,8.336599,"{""Vector"":[-0.06181473,-0.044399727,0.01341511,0.21191439,-0.10728667,-0.23058775,0.45198748,-1.1380726,-1.1180682,-0.32247138,-0 ...
SHAPE t2_scores_alias: [1567]

13. Bonus: operationalizing the pipeline -- real /jobs + /schedule

Everything above ran embedded, synchronously, in this process -- linal-hub is deliberately embedded-only (no linal serve subprocess, no Rust toolchain) by design, and the EMBEDDED_CONTRACT.md is explicit that background jobs/scheduling only exist on the real HTTP server. This section is a clearly-separated, optional bonus that crosses that boundary on purpose to answer directly: yes, this same DSL, unmodified, can run unattended -- by spinning up a real linal serve process (only if the compiled linal binary is actually on PATH; this gracefully skips otherwise) and driving its real /jobs and /schedule HTTP endpoints with nothing beyond the Python standard library (no new dependency for one bonus section).

The persisted runs dataset from §12 is copied into the server's own data directory and LOAD DATASETed there -- the server is a separate OS process with its own storage, not a shared memory space with the embedded db above.

import os

linal_binary = shutil.which("linal")
if linal_binary is None:
    print("`linal` binary not found on PATH -- skipping the real-server bonus section. "
          "Every result above already stands on its own via the embedded package.")
else:
    print(f"found real linal binary: {linal_binary}")
    server_dir = os.path.abspath("./data_manufacturing_quality_server")
    shutil.rmtree(server_dir, ignore_errors=True)
    os.makedirs(server_dir, exist_ok=True)
    shutil.copytree("./data_manufacturing_quality/default", f"{server_dir}/data/default")

    port = 18777
    proc = subprocess.Popen(
        [linal_binary, "serve", "--port", str(port)],
        cwd=server_dir,
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
    )

    def http(method, path, body=None, content_type="text/plain"):
        url = f"http://localhost:{port}{path}"
        data = body.encode() if isinstance(body, str) else (json.dumps(body).encode() if body is not None else None)
        req = urllib.request.Request(url, data=data, method=method,
                                      headers={"Content-Type": content_type} if data else {})
        with urllib.request.urlopen(req, timeout=5) as resp:
            return json.loads(resp.read())

    for _ in range(20):
        try:
            print(http("GET", "/health"))
            break
        except (urllib.error.URLError, ConnectionRefusedError):
            time.sleep(0.5)

    print(http("POST", "/execute?format=json", "LOAD DATASET runs"))
    print(http("POST", "/execute?format=json", "SELECT COUNT(*) AS n FROM runs"))

found real linal binary: /private/tmp/claude-501/-Users-nicolasbalaguera-dev-linaldb/eac69ab5-c1db-428b-8849-b5f7697586ad/scratchpad/linal_082/linal-0.1.82-aarch64-apple-darwin/linal

{'status': 'ok'}
{'status': 'ok', 'result': {'Message': "Loaded dataset 'runs' from './data/default' (1567 rows, indices restored on: features)"}}
{'status': 'ok', 'result': {'Table': {'id': 0, 'schema': {'fields': [{'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}], 'field_indices': {'n': 0}}, 'rows': [{'schema': {'fields': [{'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}], 'field_indices': {'n': 0}}, 'values': [{'Int': 1567}]}], 'metadata': {'name': 'Query Result', 'created_at': '2026-09-16T13:47:01.190125Z', 'updated_at': '2026-09-16T13:47:01.190126Z', 'version': 1, 'row_count': 1, 'column_stats': {'n': {'value_type': 'Int', 'null_count': 0, 'min': {'Int': 1567}, 'max': {'Int': 1567}}}, 'schema': {'fields': [{'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}], 'field_indices': {'n': 0}}, 'extra': {}}}}}

if linal_binary is not None:
    nightly_scan = (
        "SELECT batch_id, COUNT(*) AS n, AVG(label) AS fail_rate FROM runs "
        "GROUP BY batch_id HAVING AVG(label) > 0.15 AND COUNT(*) >= 5 "
        "ORDER BY fail_rate DESC LIMIT 5"
    )

    print("--- submit the nightly quality scan as a background job ---")
    job = http("POST", "/jobs", nightly_scan)
    print(job)
    job_id = job["job_id"]
    for _ in range(10):
        status = http("GET", f"/jobs/{job_id}")
        if status["job"]["status"] in ("Completed", "Failed"):
            break
        time.sleep(0.3)
    print(status)
    print(http("GET", f"/jobs/{job_id}/result"))

    print("\n--- register it as a real recurring scheduled task (every 5s, for this demo) ---")
    sched = http("POST", "/schedule",
                  {"name": "nightly_quality_scan", "command": nightly_scan, "interval_secs": 5},
                  content_type="application/json")
    print(sched)
    schedule_id = sched["id"]

    time.sleep(11)
    tasks = http("GET", "/schedule")
    print(tasks)
    last_run = tasks["tasks"][0]["last_run"]
    assert last_run is not None, "the scheduled task should have fired at least once by now"
    print(f"\nPASS: the scheduled task fired on its own (last_run={last_run}) -- "
          "real background scheduling, same unmodified DSL as the rest of this notebook.")

    print(http("DELETE", f"/schedule/{schedule_id}"))
    proc.terminate()
    proc.wait(timeout=5)
    print("server stopped, cleaned up.")

--- submit the nightly quality scan as a background job ---
{'status': 'ok', 'job_id': '56620d92-4121-46fe-96b6-54e285ecf23b'}

{'status': 'ok', 'job': {'id': '56620d92-4121-46fe-96b6-54e285ecf23b', 'command': 'SELECT batch_id, COUNT(*) AS n, AVG(label) AS fail_rate FROM runs GROUP BY batch_id HAVING AVG(label) > 0.15 AND COUNT(*) >= 5 ORDER BY fail_rate DESC LIMIT 5', 'status': 'Completed', 'created_at': '2026-09-16T13:47:01.198259Z', 'started_at': '2026-09-16T13:47:01.198325Z', 'finished_at': '2026-09-16T13:47:01.199390Z', 'result': {'Table': {'id': 0, 'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'rows': [{'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-08-10'}, {'Int': 12}, {'Float': 0.4166666567325592}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-07-30'}, {'Int': 5}, {'Float': 0.4000000059604645}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-07-29'}, {'Int': 12}, {'Float': 0.3333333432674408}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-08-17'}, {'Int': 17}, {'Float': 0.29411765933036804}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-07-28'}, {'Int': 8}, {'Float': 0.25}]}], 'metadata': {'name': 'Query Result', 'created_at': '2026-09-16T13:47:01.199368Z', 'updated_at': '2026-09-16T13:47:01.199368Z', 'version': 1, 'row_count': 5, 'column_stats': {'fail_rate': {'value_type': 'Float', 'null_count': 0, 'min': {'Float': 0.25}, 'max': {'Float': 0.4166666567325592}}, 'batch_id': {'value_type': 'String', 'null_count': 0, 'min': {'String': '2008-07-28'}, 'max': {'String': '2008-08-17'}}, 'n': {'value_type': 'Int', 'null_count': 0, 'min': {'Int': 5}, 'max': {'Int': 17}}}, 'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'extra': {}}}}, 'error': None, 'target_db': None}}
{'status': 'ok', 'result': {'Table': {'id': 0, 'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'rows': [{'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-08-10'}, {'Int': 12}, {'Float': 0.4166666567325592}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-07-30'}, {'Int': 5}, {'Float': 0.4000000059604645}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-07-29'}, {'Int': 12}, {'Float': 0.3333333432674408}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-08-17'}, {'Int': 17}, {'Float': 0.29411765933036804}]}, {'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'values': [{'String': '2008-07-28'}, {'Int': 8}, {'Float': 0.25}]}], 'metadata': {'name': 'Query Result', 'created_at': '2026-09-16T13:47:01.199368Z', 'updated_at': '2026-09-16T13:47:01.199368Z', 'version': 1, 'row_count': 5, 'column_stats': {'fail_rate': {'value_type': 'Float', 'null_count': 0, 'min': {'Float': 0.25}, 'max': {'Float': 0.4166666567325592}}, 'batch_id': {'value_type': 'String', 'null_count': 0, 'min': {'String': '2008-07-28'}, 'max': {'String': '2008-08-17'}}, 'n': {'value_type': 'Int', 'null_count': 0, 'min': {'Int': 5}, 'max': {'Int': 17}}}, 'schema': {'fields': [{'name': 'batch_id', 'value_type': 'String', 'nullable': False, 'is_lazy': False}, {'name': 'n', 'value_type': 'Int', 'nullable': False, 'is_lazy': False}, {'name': 'fail_rate', 'value_type': 'Float', 'nullable': False, 'is_lazy': False}], 'field_indices': {'batch_id': 0, 'fail_rate': 2, 'n': 1}}, 'extra': {}}}}}

--- register it as a real recurring scheduled task (every 5s, for this demo) ---
{'status': 'ok', 'id': '842d5236-aea6-4342-a5a9-870c59969e2c'}

{'status': 'ok', 'tasks': [{'id': '842d5236-aea6-4342-a5a9-870c59969e2c', 'name': 'nightly_quality_scan', 'command': 'SELECT batch_id, COUNT(*) AS n, AVG(label) AS fail_rate FROM runs GROUP BY batch_id HAVING AVG(label) > 0.15 AND COUNT(*) >= 5 ORDER BY fail_rate DESC LIMIT 5', 'interval_secs': 5, 'target_db': None, 'last_run': '2026-09-16T13:47:11.671841Z'}]}

PASS: the scheduled task fired on its own (last_run=2026-09-16T13:47:11.671841Z) -- real background scheduling, same unmodified DSL as the rest of this notebook.
{'status': 'ok', 'message': 'Task removed'}
server stopped, cleaned up.

Summary

Built a complete, real data-scientist pipeline on real, messy semiconductor-fab telemetry (UCI SECOM), end to end in linaldb alone:

• Feature triage of 590 real sensors down to a numerically healthy working set (documented, not hidden), motivated by a real near-singular covariance matrix at the naive top-15 pick.
• Multivariate statistical process control (Hotelling's T-squared) via real TRANSPOSE/ MATMUL/SCALE/DETERMINANT/INVERSE, cross-checked against numpy, with an honest (modest) detection-quality report.
• Virtual metrology regression solved via SOLVE, cross-checked via QR/LU decomposition reconstructions -- the first notebook to exercise any of INVERSE/SOLVE/QR/LU/ DETERMINANT.
• Relational analytics reusing GROUP BY/window functions from earlier notebooks alongside genuinely new-to-this-project-history HAVING, LAG/LEAD, and NULLIF.
• Similarity search (CREATE VECTOR INDEX/SEARCH) reused from notebook 09, applied to a new root-cause-triage use case.
• Named, chained pipelines (DEFINE/APPLY PIPELINE) -- exercised by a notebook for the first time.
• Governance: ATTACH/AUDIT DATASET (first notebook use), EXPLAIN LINEAGE, EXPORT.
• Bonus: real background job + recurring schedule against an actual linal serve process -- directly answering whether pipeline scheduling can be shown in a notebook (yes, as a clearly separated, optional, gracefully-skipping section).

Two real engine bugs found, fixed, and shipped in the same session:

1. LET <new_name> = <existing_tensor_name> (a bare-identifier RHS) silently failed to create the new binding -- no error, and the success message even named the old variable ("Defined variable: <old_name>" instead of <new_name>). Fixed by aliasing through the engine's existing bind_resource primitive (the same one BIND already used); LAZY LET/ DERIVE on a bare identifier are now clear errors instead of silent wrong successes. Also closed a related, previously-unnoticed gap in BIND itself for the LET x = dataset("foo") pattern used in §12/managed_service_demo.lnl.
2. EXPORTing a dataset with a Vector/Matrix column to CSV failed outright (Arrow's CSV writer has no FixedSizeList support) -- not previously documented as a constraint. Fixed by always using the existing JSON-string fallback encoding for CSV export of these columns, leaving SAVE DATASET/Parquet's native encoding untouched.

Shipped as linal-db-rs PR #101 (both fixes, one combined PR, three commits), released as real linaldb 0.1.8 on PyPI via PR #102 + the linaldb-v0.1.8 tag. This entire notebook was then re-run end to end against that real published wheel (§12's EXPORT runs and LET-alias cells demonstrate both fixes directly, no workaround needed) -- zero errors, zero regressions across all 10 linal-hub notebooks.