Compose reusable data conditions

// Run from the project root with the updated native development runtime.
// The seven synthetic records include missing values and exact IDs above 2^53.
import aner.data
import aner.units

var schema = data.schema()
schema = schema.column("row_id", "Int64", false)
schema = schema.column("age", "Int64", true)
schema = schema.column("score", "Float64", true)
schema = schema.column("reference", "Float64", false)
schema = schema.column("cohort", "String", false)
schema = schema.column("active", "Bool", false)
schema = schema.unit_column("temperature", "degC", true)
let source = data.scan_csv("examples/fixtures/data_filtering.csv", schema, "NA")

// Columns and conditions are immutable, reusable native expressions.
let age = data.col("age")
let score = data.col("score")
let population = age.between(18, 65) &&
    (data.col("cohort").is_in(List("A", "B")) || (data.col("active") == true))
let improving = score > data.col("reference")
let eligible = population && improving
let selected = source.filter(eligible)
    .filter(data.col("temperature") >= 21.0@degC)
    .select(List("row_id", "score", "temperature"))
    .collect(max_rows: 0, max_bytes: 1048576, batch_rows: 2)
print("Selected rows and exact observation IDs")
print(selected.rows())
print(selected.integer(0, "row_id"))
print(selected.integer(1, "row_id"))
print("Mean selected temperature in Celsius")
print(selected.scan_table().summarize(1048576, 2).unit_mean("temperature", "degC").value())

// Table operations return lazy snapshot scans; the original seven rows remain.
let table = source.collect(0, 1048576, 2)
let derived = table.filter(eligible).select(List("row_id", "score"))
    .collect(0, 1048576, 1)
print("Original and derived row counts")
print(table.rows())
print(derived.rows())

// Frame views keep exact observation IDs and their scientific field context.
let observations = data.frame(table, "row_id", "cohort", "reference")
let view = observations.filter(eligible).select(List("score", "temperature"))
print("Frame source rows, view rows and retained observation IDs")
print(observations.rows())
print(view.rows())
print(view.row_id(0))
print(view.row_id(1))

// NOT keeps unknown missing comparisons unknown, rather than turning them true.
print("Rows outside the range, then rows outside or missing")
print(source.filter(!age.between(18, 65)).summarize(1048576, 2).rows())
print(source.filter(age.is_missing() || !age.between(18, 65))
    .summarize(1048576, 2).rows())
print("Rows with a present score")
print(source.filter(score.is_present()).summarize(1048576, 2).rows())

// Plans preserve execution budgets and spill policy. This fixture fits RAM;
// larger inputs transparently spill the retained selected values when needed.
let bounded = source.with_spill("", 1073741824).filter(eligible)
    .bounded(max_scanned_rows: 100000, max_read_bytes: 16777216, timeout_ms: 1000)
    .preview(max_rows: 10, max_bytes: 131072, batch_rows: 64)
print("Preview row count and full-source validation status")
print(bounded.rows())
print(bounded.complete())
// Persist data.provenance(selected) with research artifacts when appropriate.