Data analysis (tables)

-- Doc example: data analysis on tables (lists of maps) — load typed CSV and Parquet, decide what
-- missing means, join, summarize, pivot, then prove which data produced the result (lineage + receipt).
intent: "doc example: data analysis"
require file("_doctest_analysis_sales.csv")
require file("_doctest_analysis_regions.parquet")
require sign("ANALYST")

-- The inputs. A real pipeline reads files someone else wrote; here the program writes them first so
-- the example is self-contained.
write_file("_doctest_analysis_sales.csv", "order,day,region,amount\n1,2026-01-05,north,120.50\n2,2026-01-06,south,80\n3,2026-01-19,north,\n4,2026-02-02,west,150.25\n5,2026-02-03,north,200\n6,2026-02-10,south,NA\n")
write_file("_doctest_analysis_regions.parquet", parquet_write([{"region": "north", "manager": "Ana"}, {"region": "south", "manager": "Luis"}, {"region": "east", "manager": "Mei"}]))

-- 1. Load with types: an empty field (and the NA mark) is nothing = missing
let sales be csv_parse(read_file("_doctest_analysis_sales.csv"), {"types": {"order": "int", "day": "date", "amount": "decimal"}, "missing": ["NA"]})
let regions be parquet_read(read_file_bytes("_doctest_analysis_regions.parquet"))

-- 2. Decide what missing means: an order without an amount is dropped
let clean be drop_missing(sales, "amount")

-- 3. Join (a region with no match gets nothing, then a label), then aggregate one row per region
let with_manager be fill_missing(join(clean, regions, "region", "left"), {"manager": "unassigned"})
let by_region be summarize(with_manager, "region", {"total": sum_of("amount"), "orders": count(), "avg": mean_of("amount"), "manager": first_of("manager")})
let ranked be sort_by(by_region, (r) => r.total, desc = true)

-- 4. Region x month
let monthly be pivot(apply(clean, (r) => merge(r, {"month": format_time(truncate(r.day, "month"), "%Y-%m")})), "region", "month", "amount", sum_of("amount"))

each r in ranked
    print(`{r.region}: {r.total} in {r.orders} orders ({r.manager})`)

-- 5. Prove it: which inputs, which program, which result
let key be as_secret(bytes("0707070707070707070707070707070707070707070707070707070707070707", "hex"), "ANALYST")
let did be did_key_encode(ed25519_pubkey(key))
let proof be receipt({"sign": key, "verification_method": did + "#" + did_key_decode(did).multibase, "result": ranked})

test "load: typed columns, and missing values are nothing"
    assert_eq(length(sales), 6)
    assert_eq(sales[0].day, date(2026, 1, 5))
    assert_eq(sales[0].amount, 120.50d)
    assert_eq(sales[2].amount, nothing)                                   -- empty field
    assert_eq(sales[5].amount, nothing)                                   -- the NA mark
    assert_eq(count_missing(collect(sales, "amount")), 2)
    assert_eq(count(collect(sales, "amount")), 4)                         -- present values
    assert_eq(regions[0], {"region": "north", "manager": "Ana"})

test "summarize: one row per group, aggregates skip nothing"
    assert_eq(ranked[0], {"region": "north", "total": 320.50d, "orders": 2, "avg": 160.25d, "manager": "Ana"})
    assert_eq(collect(ranked, "region"), ["north", "west", "south"])
    assert_eq(join(clean, regions, "region", "left")[2].manager, nothing)  -- west has no match in regions
    assert_eq(ranked[1].manager, "unassigned")                            -- filled on purpose
    -- without dropping first, the aggregates skip the missing amounts but count() counts rows
    let raw be summarize(sales, "region", {"total": sum_of("amount"), "n": count(), "priced": (g) => count(collect(g, "amount"))})
    assert_eq(raw[0], {"region": "north", "total": 320.50d, "n": 3, "priced": 2})

test "every aggregate"
    let rows be [{"r": "n", "m": 10}, {"r": "s", "m": 5}, {"r": "n", "m": 1}, {"r": "s", "m": nothing}]
    let s be summarize(rows, "r", {"sum": sum_of("m"), "mean": mean_of("m"), "min": min_of("m"), "max": max_of("m"), "median": median_of("m"), "q": quantile_of("m", 0.5), "first": first_of("m"), "distinct": n_unique_of("m"), "rows": count(), "span": (g) => max(collect(g, "m")) - min(collect(g, "m"))})
    assert_eq(s[0], {"r": "n", "sum": 11, "mean": 5.5, "min": 1, "max": 10, "median": 5.5, "q": 5.5, "first": 10, "distinct": 2, "rows": 2, "span": 9})
    assert_eq(s[1].first, 5)
    -- by several columns (each key column kept) or by a function (the key lands in "key")
    let two be summarize([{"a": 1, "b": "x", "v": 1}, {"a": 1, "b": "y", "v": 2}, {"a": 1, "b": "x", "v": 3}], ["a", "b"], {"s": sum_of("v")})
    assert_eq(two, [{"a": 1, "b": "x", "s": 4}, {"a": 1, "b": "y", "s": 2}])
    let by_month be summarize(clean, (r) => truncate(r.day, "month"), {"total": sum_of("amount")})
    assert_eq(by_month, [{"key": date(2026, 1, 1), "total": 200.50d}, {"key": date(2026, 2, 1), "total": 350.25d}])
    -- a group whose values are all missing: sum_of is 0, the others nothing
    let uneven be summarize([{"g": "a", "v": 1}, {"g": "c"}], "g", {"s": sum_of("v"), "m": mean_of("v")})
    assert_eq(uneven[1], {"g": "c", "s": 0, "m": nothing})

test "a column no row has is an error, not a silent zero"
    assert_error(() => summarize(clean, "region", {"t": sum_of("amuont")}))
    assert_error(() => group_by(clean, "regoin"))
    assert_error(() => summarize(clean, "region", {"region": count()}))  -- would overwrite the key

test "group_by returns [{key, items}] in first-appearance order; count_by counts, most frequent first"
    let groups be group_by(clean, "region")
    assert_eq(collect(groups, "key"), ["north", "south", "west"])
    assert_eq(length(groups[0].items), 2)
    assert_eq(group_by([1, 1.0, 2], (v) => v), [{"key": 1, "items": [1, 1.0]}, {"key": 2, "items": [2]}])
    assert_eq(count_by(sales, "region"), [{"key": "north", "count": 3}, {"key": "south", "count": 2}, {"key": "west", "count": 1}])
    assert_eq(count_by(["x", "y", "x"]), [{"key": "x", "count": 2}, {"key": "y", "count": 1}])

test "join: inner, left, right, outer, semi, anti"
    let L be [{"id": 1, "a": "x"}, {"id": 2, "a": "y"}]
    let R be [{"id": 1, "b": "p"}, {"id": 3, "b": "q"}]
    assert_eq(join(L, R, "id"), [{"id": 1, "a": "x", "b": "p"}])                          -- inner is the default
    assert_eq(join(L, R, "id", "left"), [{"id": 1, "a": "x", "b": "p"}, {"id": 2, "a": "y", "b": nothing}])
    assert_eq(join(L, R, "id", "right"), [{"id": 1, "a": "x", "b": "p"}, {"id": 3, "a": nothing, "b": "q"}])
    assert_eq(join(L, R, "id", "outer"), [{"id": 1, "a": "x", "b": "p"}, {"id": 2, "a": "y", "b": nothing}, {"id": 3, "a": nothing, "b": "q"}])
    assert_eq(join(L, R, "id", "semi"), [{"id": 1, "a": "x"}])
    assert_eq(join(L, R, "id", "anti"), [{"id": 2, "a": "y"}])
    assert_eq(join([{"a": 1, "b": 1, "v": "x"}], [{"a": 1, "b": 1, "w": "y"}], ["a", "b"]), [{"a": 1, "b": 1, "v": "x", "w": "y"}])
    assert_eq(join(L, [{"id": 1, "a": "z"}], "id", "left"), [{"id": 1, "a": "x", "a_right": "z"}, {"id": 2, "a": "y", "a_right": nothing}])
    assert_error(() => join([{"k": 1, "b": 1, "b_right": 2}], [{"k": 1, "b": 3}], "k"))  -- b_right is taken
    assert_eq(join([{"k": nothing, "a": 1}], [{"k": nothing, "b": 2}], "k"), [])          -- a nothing key never matches
    assert_eq(join(["a", "b"], "-"), "a-b")                                               -- two arguments: the TEXT join

test "pivot: one row per index value, one column per distinct value"
    assert_eq(monthly, [{"region": "north", "2026-01": 120.50d, "2026-02": 200d}, {"region": "south", "2026-01": 80d, "2026-02": nothing}, {"region": "west", "2026-01": nothing, "2026-02": 150.25d}])
    let P be [{"d": "mon", "p": "a", "v": 1}, {"d": "mon", "p": "b", "v": 2}, {"d": "tue", "p": "a", "v": 3}]
    assert_eq(pivot(P, "d", "p", "v"), [{"d": "mon", "a": 1, "b": 2}, {"d": "tue", "a": 3, "b": nothing}])
    assert_error(() => pivot(P + [{"d": "mon", "p": "a", "v": 10}], "d", "p", "v"))      -- two rows in one cell: pass agg
    assert_eq(pivot(P + [{"d": "mon", "p": "a", "v": 10}], "d", "p", "v", sum_of("v"))[0].a, 11)
    assert_error(() => pivot([{"d": 1, "c": 1, "v": 1}, {"d": 1, "c": "1", "v": 1}], "d", "c", "v"))   -- 1 and "1" → one column name

test "missing data: nothing is missing, NaN is invalid"
    assert_eq(is_missing(nothing), true)
    assert_eq(fill_missing([1, nothing], 0), [1, 0])
    assert_eq(fill_missing([{"a": nothing, "b": nothing}], 0), [{"a": 0, "b": 0}])
    assert_eq(fill_missing([{"a": nothing, "b": nothing}], {"a": 1}), [{"a": 1, "b": nothing}])
    assert_eq(drop_missing([1, nothing, 2]), [1, 2])
    assert_eq(drop_missing([{"a": 1, "b": nothing}, {"a": 2, "b": 3}]), [{"a": 2, "b": 3}])     -- any column
    assert_eq(drop_missing([{"a": 1, "b": nothing}, {"a": nothing, "b": 2}], "a"), [{"a": 1, "b": nothing}])
    assert_eq(drop_missing([{"a": 1, "b": nothing}, {"a": 1}], ["a", "b"]), [])            -- an absent column counts as missing
    assert_eq(fill_nan([nan, 1], 0), [0, 1])
    assert_eq(mean([1, nothing, 3]), 2.0)
    assert(is_nan(mean([1, nan, 3])))

test "lineage: every input the program read, with its sha256"
    let l be lineage()
    assert_eq(collect(l, "source"), ["read_file", "read_file_bytes", "parquet_read"])
    assert_eq(l[0].what, "_doctest_analysis_sales.csv")
    assert_eq(l[0].encoding, "text")
    assert_eq("0x" + l[0].sha256, hex(sha256(read_file("_doctest_analysis_sales.csv"))))   -- anyone can recompute it
    assert_eq(l[1].encoding, "bytes")
    assert_eq(l[2].what, "bytes " + text(l[1].bytes))                      -- a file passed as bytes: its size, never its content
    assert_eq("0x" + l[2].sha256, hex(sha256(canonical_json(regions))))    -- encoding "jcs"

test "receipt: the lineage travels as inputs, signed, next to the program and the result"
    assert_eq(length(proof.credentialSubject.inputs), 3)
    assert_eq(proof.credentialSubject.inputs[0].sha256, lineage()[0].sha256)
    assert_eq(proof.credentialSubject.declared_result_sha256, slice(hex(sha256(canonical_json(ranked))), 2))
    assert_eq(proof.issuer, did)
    assert_eq(receipt_verify(proof, did).verified, true)
    let forged be merge(proof, {"credentialSubject": merge(proof.credentialSubject, {"inputs": []})})
    assert_eq(receipt_verify(forged, did), nothing)                        -- drop an input and the proof breaks

test "keys group by ==: value and type kept, first-appearance order"
    assert_eq(count_by([1, 1.0, true, "1"]), [{"key": 1, "count": 3}, {"key": "1", "count": 1}])
    assert_eq(count_by([{"a": 1, "b": 2}, {"b": 2, "a": 1}])[0].count, 2)     -- same entries, another order
    assert_eq(count_by([datetime("2026-01-01T12:00:00Z"), datetime("2026-01-01T09:00:00-03:00")])[0].count, 2)   -- same instant
    assert_eq(count_by([nan, nan, 1])[0].count, 2)                            -- every NaN in one group
    assert_error(() => count_by([1.5d, 1.5]))                                 -- decimal vs float: convert on purpose
    let span be summarize([{"g": "a", "d": date(2026, 1, 2)}, {"g": "a", "d": date(2026, 1, 1)}], "g", {"first": min_of("d"), "last": max_of("d")})
    assert_eq(span, [{"g": "a", "first": date(2026, 1, 1), "last": date(2026, 1, 2)}])
    assert_error(() => summarize([{"g": "a", "v": "x"}], "g", {"s": sum_of("v")}))   -- text in a numeric aggregate
    let by_pair be group_by([{"a": 1, "b": 2, "v": 1}, {"a": 1, "b": 2, "v": 2}], ["a", "b"])
    assert_eq(by_pair[0].key, {"a": 1, "b": 2})                               -- a list of columns: the key is a map

test "rows are values: each gives a copy, so build the new table"
    let rows be [{"x": 1}, {"x": 2}]
    each r in rows
        set r["x"] to 0
    assert_eq(rows, [{"x": 1}, {"x": 2}])                                     -- unchanged
    set rows to apply(rows, (r) => merge(r, {"y": r.x * 10}))
    assert_eq(rows, [{"x": 1, "y": 10}, {"x": 2, "y": 20}])
    assert_eq(where(rows, (r) => r.x > 1), [{"x": 2, "y": 20}])
    assert_eq(collect(rows, "y"), [10, 20])
    assert_eq(sort_by(rows, (r) => r.y, desc = true)[0].x, 2)