Example: JOIN Queries with Aggregations¶
Multi-table JOINs and GROUP BY aggregations are the patterns most common in analytics and reporting — and where CIUF's incremental model saves the most work.
Why joins and aggregations benefit most¶
CIUF's DAG model stores each table's data at the TableNode level. When rows change, the delta propagates through the JOIN and GROUP BY nodes without reloading anything from PostgreSQL. For queries that run hundreds of times with rare writes, this eliminates essentially all database load.
The benefit scales with:
- Repetition rate — how many times the same query runs before the data changes
- Join cardinality — larger joins mean more computation saved per cache hit
- Aggregation depth — deep GROUP BY + aggregate chains amplify the savings
Schema¶
CREATE TABLE regions (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL
);
CREATE TABLE stores (
id SERIAL PRIMARY KEY,
region_id INT REFERENCES regions(id),
name TEXT NOT NULL
);
CREATE TABLE products (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
category TEXT NOT NULL
);
CREATE TABLE orders (
id SERIAL PRIMARY KEY,
store_id INT REFERENCES stores(id),
product_id INT REFERENCES products(id),
quantity INT NOT NULL,
amount NUMERIC(12,2) NOT NULL,
status TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
Example 1 — Two-table JOIN with WHERE¶
The simplest cached join: orders enriched with product name, filtered by status.
from ciuf import Engine, from_sql
engine = Engine("postgresql://user:pass@localhost/mydb")
result = from_sql(engine, """
SELECT
orders.id,
orders.amount,
orders.quantity,
orders.created_at,
products.name AS product_name,
products.category AS product_category
FROM orders
JOIN products ON orders.product_id = products.id
WHERE orders.status = 'completed'
ORDER BY orders.created_at DESC
LIMIT 500
""")
df = result.query() # first call: PostgreSQL
df = result.query() # subsequent calls: in-process cache
print(df.dtypes)
Example 2 — Three-table JOIN with GROUP BY and aggregations¶
Revenue breakdown by region and product category — a typical dashboard query.
result = from_sql(engine, """
SELECT
regions.name AS region,
products.category,
SUM(orders.amount) AS total_revenue,
COUNT(*) AS order_count,
AVG(orders.amount) AS avg_order_value,
MIN(orders.created_at) AS first_order,
MAX(orders.created_at) AS last_order
FROM orders
JOIN stores ON orders.store_id = stores.id
JOIN regions ON stores.region_id = regions.id
JOIN products ON orders.product_id = products.id
WHERE orders.status = 'completed'
GROUP BY regions.name, products.category
ORDER BY total_revenue DESC
""")
df = result.query()
print(df.head(10))
CIUF builds the DAG:
TableNode(orders) → JoinNode(stores) → JoinNode(regions) → JoinNode(products)
→ GroupNode(region, category)
→ SelectNode
Each GroupNode maintains accumulators (SUM, COUNT, AVG, MIN, MAX) per group key. When an order is inserted, only the affected group updates.
Example 3 — Incremental update on write¶
This shows the full read-write cycle: query once, write, query again — same result object, updated automatically.
import time
engine = Engine("postgresql://user:pass@localhost/mydb")
result = from_sql(engine, """
SELECT
stores.name AS store,
products.category,
SUM(orders.amount) AS revenue,
COUNT(*) AS orders
FROM orders
JOIN stores ON orders.store_id = stores.id
JOIN products ON orders.product_id = products.id
WHERE orders.status = 'completed'
GROUP BY stores.name, products.category
ORDER BY revenue DESC
""")
# Cold read
df_before = result.query()
print("Before insert:")
print(df_before.head())
# Simulate an order insert (in real code this comes from your write path)
engine.on_insert("orders", {
"id": 99999,
"store_id": 1,
"product_id": 3,
"quantity": 2,
"amount": 299.98,
"status": "completed",
"created_at": "2026-05-10T14:00:00+00:00",
})
# Hot read — reflects the new order, no PostgreSQL query
t0 = time.perf_counter()
df_after = result.query()
print(f"\nAfter insert ({(time.perf_counter()-t0)*1000:.3f} ms):")
print(df_after.head())
The revenue and orders columns for store 1 / category of product 3 reflect the new order. The update took microseconds because only one group key was touched.
Example 4 — Benchmark harness¶
Measure the exact speedup on your own data:
import statistics
import time
from ciuf import Engine, from_sql
engine = Engine("postgresql://user:pass@localhost/mydb")
SQL = """
SELECT
regions.name AS region,
products.category,
SUM(orders.amount) AS revenue,
COUNT(*) AS orders,
AVG(orders.amount) AS aov
FROM orders
JOIN stores ON orders.store_id = stores.id
JOIN regions ON stores.region_id = regions.id
JOIN products ON orders.product_id = products.id
WHERE orders.status = 'completed'
GROUP BY regions.name, products.category
"""
result = from_sql(engine, SQL)
# Cold read
t0 = time.perf_counter()
df = result.query()
cold_ms = (time.perf_counter() - t0) * 1000
# 100 hot reads
hot_times = []
for _ in range(100):
t0 = time.perf_counter()
result.query()
hot_times.append((time.perf_counter() - t0) * 1000)
p50 = statistics.median(hot_times)
p99 = sorted(hot_times)[98]
print(f"Rows returned : {len(df)}")
print(f"Cold read : {cold_ms:.1f} ms")
print(f"Hot p50 : {p50:.3f} ms")
print(f"Hot p99 : {p99:.3f} ms")
print(f"Speedup (p50) : {cold_ms / p50:.0f}×")
Typical output on a three-way JOIN, 250k orders, 20 result rows:
Supported aggregation functions¶
| Function | Supported |
|---|---|
COUNT(*) |
✅ |
COUNT(col) |
✅ |
SUM(col) |
✅ |
AVG(col) |
✅ |
MIN(col) |
✅ |
MAX(col) |
✅ |
Window functions (RANK, ROW_NUMBER) |
❌ v1 scope |
HAVING |
❌ v1 scope |
Cache invalidation for aggregations¶
When CIUF receives a write event, the GroupNode recomputes only the affected group keys:
INSERT into orders (store_id=1, product_id=3, amount=299.98)
→ GroupNode: update accumulators for (region_of_store_1, category_of_product_3)
→ SelectNode: re-project the updated group rows
The cost is proportional to the number of affected groups, not the total number of groups. Aggregating over millions of rows costs the same per-write as aggregating over thousands, as long as writes touch a small number of groups at a time.