fix(inputlayer-py): compile each query to one program with columns by position - #235

    Finished `release` profile [optimized] target(s) in 0.18s
Resolved 67 packages in 1ms
Checked 63 packages in 0.79ms
============================= test session starts ==============================
platform linux -- Python 3.10.22, pytest-9.1.1, pluggy-1.6.0 -- ~/.no-mistakes/worktrees/9eea230c3aae/01M43B829RPSFJPWZ29ZD3GEET/packages/inputlayer-py/.venv/bin/python
cachedir: .pytest_cache
rootdir: ~/.no-mistakes/worktrees/9eea230c3aae/01M43B829RPSFJPWZ29ZD3GEET/packages/inputlayer-py
configfile: pyproject.toml
plugins: langsmith-0.12.1, asyncio-1.3.0, anyio-4.13.0
asyncio: mode=auto, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collecting ... collected 35 items

tests/docs_queries_live.py::test_basic_queries_all_rows PASSED           [  2%]
tests/docs_queries_live.py::test_basic_queries_with_a_filter PASSED      [  5%]
tests/docs_queries_live.py::test_selecting_specific_columns PASSED       [  8%]
tests/docs_queries_live.py::test_a_projection_is_a_set PASSED            [ 11%]
tests/docs_queries_live.py::test_joins PASSED                            [ 14%]
tests/docs_queries_live.py::test_self_joins PASSED                       [ 17%]
tests/docs_queries_live.py::test_computed_columns PASSED                 [ 20%]
tests/docs_queries_live.py::test_ordering_and_pagination PASSED          [ 22%]
tests/docs_queries_live.py::test_sort_and_limit_run_in_the_engine PASSED [ 25%]
tests/docs_queries_live.py::test_aggregations_group_by_department_with_stats PASSED [ 28%]
tests/docs_queries_live.py::test_agg_two_measures_same_column PASSED     [ 31%]
tests/docs_queries_live.py::test_aggregates_leave_no_session_rule PASSED [ 34%]
tests/docs_queries_live.py::test_count_without_a_column PASSED           [ 37%]
tests/docs_queries_live.py::test_aggregations_top_k_per_group PASSED     [ 40%]
tests/docs_queries_live.py::test_aggregate_ordered_and_limited PASSED    [ 42%]
tests/docs_queries_live.py::test_working_with_results PASSED             [ 45%]
tests/docs_queries_live.py::test_query_plans PASSED                      [ 48%]
tests/docs_queries_live.py::test_debug_takes_the_arguments_of_query PASSED [ 51%]
tests/docs_queries_live.py::test_debug_shows_the_plan_of_an_aggregate_query PASSED [ 54%]
tests/docs_queries_live.py::test_why_returns_the_selected_columns_with_a_proof_per_row PASSED [ 57%]
tests/docs_queries_live.py::test_why_explains_an_aggregate_query PASSED  [ 60%]
tests/docs_queries_live.py::test_why_returns_a_column_of_the_second_joined_relation PASSED [ 62%]
tests/docs_queries_live.py::test_why_returns_a_computed_column PASSED    [ 65%]
tests/docs_queries_live.py::test_an_offset_without_a_limit_pages_in_the_engine PASSED [ 68%]
tests/docs_queries_live.py::test_why_orders_and_paginates_like_query PASSED [ 71%]
tests/docs_queries_live.py::test_why_returns_the_distinct_rows_of_a_projection_like_query PASSED [ 74%]
tests/docs_queries_live.py::test_paging_a_projection_ordered_by_an_unselected_column_is_refused PASSED [ 77%]
tests/docs_queries_live.py::test_why_orders_and_limits_an_aggregate_query_like_query PASSED [ 80%]
tests/docs_queries_live.py::test_raw_iql PASSED                          [ 82%]
tests/docs_queries_live.py::test_derived_relations_querying_a_recursive_rule PASSED [ 85%]
tests/docs_queries_live.py::test_sessions_session_facts_mix_with_persistent_data PASSED [ 88%]
tests/docs_queries_live.py::test_session_insert_keeps_every_fact PASSED  [ 91%]
tests/docs_queries_live.py::test_distance_functions PASSED               [ 94%]
tests/docs_queries_live.py::test_temporal_functions PASSED               [ 97%]
tests/docs_queries_live.py::test_or_conditions_merge_their_branches PASSED [100%]

============================== 35 passed in 6.52s ==============================

=== avg+max salary by department (bug: dropped groups)
  programs sent: 1
  >>> il_q(Department, avg<Salary>, max<Salary>) <- staff(Id, Department, Salary)
      ?il_q(Department, AvgSalary, MaxSalary)
  columns: ['department', 'avg_salary', 'max_salary'] rows: [('eng', 106.66666666666667, 120.0), ('hr', 90.0, 90.0)]

=== count() without a column
  programs sent: 1
  >>> il_q(count<Id>) <- emp(Id, Name, Department, Salary)
      ?il_q(Count)
  columns: ['count'] scalar: 5

=== projection Emp.department (F8: projection, set)
  programs sent: 1
  >>> ?emp(Id, Name, Department, Salary)
  columns: ['department'] rows: [('eng',), ('hr',)]

=== projection Emp.name, Emp.salary order desc limit 2 offset 1
  programs sent: 1
  >>> il_q(Name, Salary) <- emp(Id, Name, Department, Salary)
      ?il_q(Name, Salary:desc), limit(2, 1)
  rows: [('Alice', 120.0), ('Cara', 110.0)]

=== offset without limit (engine-side paging)
  programs sent: 1
  >>> il_q(Name, Salary) <- emp(Id, Name, Department, Salary)
      ?il_q(Name, Salary:desc), limit(9223372036854775807, 3)
  rows: [('Eve', 95.0), ('Bob', 90.0)]

=== ADVERSARIAL: paged projection ordered by unselected column
  CompileError: ordering by a column you do not select with limit is ambiguous; select it or drop limit (add the order_by column to the selection, or drop limit and offset)
  programs sent: 0

=== debug(join=...) (F9: used to raise TypeError)
  programs sent: 1
  >>> .debug ?emp(Id, Name, Department, Salary), Salary > 100
  iql: ?emp(Id, Name, Department, Salary), Salary > 100

=== why lossy projection paged (dedupe like query)
  programs sent: 1
  >>> .why il_q(Department, Id, Name, Salary) <- emp(Id, Name, Department, Salary)
  rows: [('eng',), ('hr',)] proofs: 2

=== derived relation named Order (row attrs, no engine var names)
  programs sent: 1
  >>> ?order(Src, Dst)
  columns: ['src', 'dst'] dicts: [{'src': 1, 'dst': 2}, {'src': 2, 'dst': 3}]

=== 100 aggregate queries: programs sent 100 (one per call)
  .session -> [['No session data defined.']]

=== session.insert two facts
  programs sent: 2
  >>> emp(10, "S1", "tmp", 1.0)
  >>> emp(11, "S2", "tmp", 2.0)
  session facts visible: [('S1',), ('S2',)]

=== latency over 50 paged queries: p50 1.30 ms, p95 1.46 ms

"""Drive the Python SDK against a live engine, logging every program sent on the wire."""
import asyncio, os, sys, time
from typing import ClassVar
from inputlayer import InputLayer, Relation, Derived, From, avg, count, max_
from inputlayer.exceptions import CompileError

URL = os.environ["INPUTLAYER_TEST_SERVER"]

class Staff(Relation):
    id: int
    department: str
    salary: float

class Emp(Relation):
    id: int
    name: str
    department: str
    salary: float

class Edge(Relation):
    src: int
    dst: int

class Order(Derived):
    src: int
    dst: int
    rules: ClassVar[list] = []
Order.rules = [From(Edge).select(src=Edge.src, dst=Edge.dst)]

SENT = []
def hook(conn):
    for name in ("_send_and_recv",):
        orig = getattr(conn, name)
        async def wrapped(prog, *a, _orig=orig, **k):
            if not prog.startswith(".kg use"):
                SENT.append(prog)
            return await _orig(prog, *a, **k)
        setattr(conn, name, wrapped)

async def call(label, coro_fn):
    SENT.clear()
    print(f"\n=== {label}")
    try:
        r = await coro_fn()
    except CompileError as e:
        print(f"  CompileError: {e}")
        print(f"  programs sent: {len(SENT)}")
        return None
    print(f"  programs sent: {len(SENT)}")
    for p in SENT:
        print("  >>> " + p.replace("\n", "\n      "))
    return r

async def main():
    il = InputLayer(URL, username="admin", password="admin"); await il.connect()
    try: await il.drop_knowledge_graph("demo_wire")
    except Exception: pass
    kg = il.knowledge_graph("demo_wire")
    hook(kg._conn)
    await kg.define(Staff, Emp, Edge)
    await kg.insert([Staff(id=1, department="eng", salary=120.0), Staff(id=2, department="eng", salary=100.0),
                     Staff(id=3, department="eng", salary=100.0), Staff(id=4, department="hr", salary=90.0)])
    await kg.insert([Emp(id=1, name="Alice", department="eng", salary=120.0), Emp(id=2, name="Bob", department="hr", salary=90.0),
                     Emp(id=3, name="Cara", department="eng", salary=110.0), Emp(id=4, name="Dana", department="eng", salary=130.0),
                     Emp(id=5, name="Eve", department="hr", salary=95.0)])
    await kg.insert([Edge(src=1, dst=2), Edge(src=2, dst=3)])

    r = await call("avg+max salary by department (bug: dropped groups)",
                   lambda: kg.query(Staff.department, avg(Staff.salary), max_(Staff.salary), join=[Staff]))
    print("  columns:", r.columns, "rows:", sorted(r.to_tuples()))

    r = await call("count() without a column", lambda: kg.query(count(), join=[Emp]))
    print("  columns:", r.columns, "scalar:", r.scalar())

    r = await call("projection Emp.department (F8: projection, set)", lambda: kg.query(Emp.department, join=[Emp]))
    print("  columns:", r.columns, "rows:", sorted(r.to_tuples()))

    r = await call("projection Emp.name, Emp.salary order desc limit 2 offset 1",
                   lambda: kg.query(Emp.name, Emp.salary, join=[Emp], order_by=Emp.salary.desc(), limit=2, offset=1))
    print("  rows:", r.to_tuples())

    r = await call("offset without limit (engine-side paging)",
                   lambda: kg.query(Emp.name, Emp.salary, join=[Emp], order_by=Emp.salary.desc(), offset=3))
    print("  rows:", r.to_tuples())

    await call("ADVERSARIAL: paged projection ordered by unselected column",
               lambda: kg.query(Emp.department, join=[Emp], order_by=Emp.salary.desc(), limit=2))

    r = await call("debug(join=...) (F9: used to raise TypeError)",
                   lambda: kg.debug(Emp.name, join=[Emp], where=lambda e: e.salary > 100))
    print("  iql:", r.iql)

    r = await call("why lossy projection paged (dedupe like query)",
                   lambda: kg.why(Emp.department, join=[Emp], order_by=Emp.department.asc(), limit=2))
    print("  rows:", r.results.to_tuples(), "proofs:", len(r.proof_trees))

    await kg.define_rules(Order)
    r = await call("derived relation named Order (row attrs, no engine var names)", lambda: kg.query(Order))
    print("  columns:", r.columns, "dicts:", sorted(r.to_dicts(), key=lambda d: d["src"]))

    SENT.clear()
    for _ in range(100):
        await kg.query(Staff.department, count(Staff.id), join=[Staff])
        await asyncio.sleep(0.02)
    print(f"\n=== 100 aggregate queries: programs sent {len(SENT)} (one per call)")
    s = await kg.execute(".session")
    print("  .session ->", s.rows)

    r = await call("session.insert two facts", lambda: kg.session.insert([Emp(id=10, name="S1", department="tmp", salary=1.0),
                                                                          Emp(id=11, name="S2", department="tmp", salary=2.0)]))
    r = await kg.query(Emp.name, join=[Emp], where=lambda e: e.department == "tmp")
    print("  session facts visible:", sorted(r.to_tuples()))

    ts = []
    for _ in range(50):
        t = time.perf_counter(); await kg.query(Emp.name, Emp.salary, join=[Emp], order_by=Emp.salary.desc(), limit=3)
        ts.append((time.perf_counter() - t) * 1000); await asyncio.sleep(0.02)
    ts.sort(); print(f"\n=== latency over 50 paged queries: p50 {ts[25]:.2f} ms, p95 {ts[47]:.2f} ms")
    await il.drop_knowledge_graph("demo_wire"); await il.close()

asyncio.run(main())

base1 programs: ['?emp(Id, Name, Department, Salary), limit(3)'] rows: [(2, 'n2', 'd', 2.0), (1, 'n1', 'd', 1.0), (0, 'n0', 'd', 0.0)]
  paged: p50 1.64 ms p95 1.94 ms
  plain: p50 1.54 ms p95 1.74 ms
target1 programs: ['il_q(Name, Salary) <- emp(Id, Name, Department, Salary)\n?il_q(Name, Salary:desc), limit(3)'] rows: [('n199', 199.0), ('n198', 198.0), ('n197', 197.0)]
  paged: p50 1.61 ms p95 1.84 ms
  plain: p50 1.26 ms p95 1.43 ms
base2 programs: ['?emp(Id, Name, Department, Salary), limit(3)'] rows: [(2, 'n2', 'd', 2.0), (1, 'n1', 'd', 1.0), (0, 'n0', 'd', 0.0)]
  paged: p50 1.22 ms p95 1.36 ms
  plain: p50 1.11 ms p95 1.30 ms
target2 programs: ['il_q(Name, Salary) <- emp(Id, Name, Department, Salary)\n?il_q(Name, Salary:desc), limit(3)'] rows: [('n199', 199.0), ('n198', 198.0), ('n197', 197.0)]
  paged: p50 1.44 ms p95 1.71 ms
  plain: p50 1.17 ms p95 1.41 ms

programs sent: 1
>>> il_q(Department, avg<Salary>, max<Salary>) <- staff(Id, Department, Salary)
?il_q(Department, AvgSalary, MaxSalary)
columns: ['department', 'avg_salary', 'max_salary'] rows: [('eng', 106.67, 120.0), ('hr', 90.0, 90.0)]