Skip to content

Commit c26bdbc

Browse files
chore: update duckdb, improve queries, add upgrade-duckdb command
Signed-off-by: Henry <mail@henrygressmann.de>
1 parent 89550dc commit c26bdbc

14 files changed

Lines changed: 585 additions & 307 deletions

File tree

‎Cargo.lock‎

Lines changed: 78 additions & 140 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎bun.lock‎

Lines changed: 88 additions & 94 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎crates/liwan/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ ua-parser = "0.2"
9393
url = "2.5"
9494

9595
# database
96-
duckdb = { version = "=1.10506", features = ["bundled", "chrono", "r2d2"] }
96+
duckdb = { version = "1.20000", features = ["bundled", "chrono", "r2d2"] }
9797
maxminddb = { version = "0.32", optional = true }
9898
mimalloc = { version = "0.1" }
9999
r2d2 = { version = "0.8", default-features = false }

‎crates/liwan/benches/reports.rs‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,17 @@ fn benchmark_reports(c: &mut Criterion) {
7373
group.finish();
7474
}
7575

76+
{
77+
let mut group = c.benchmark_group("custom_events_report");
78+
configure_group(&mut group);
79+
group.bench_function("counts", |b| {
80+
b.iter(|| {
81+
reports::custom_events_report(&conn, &entities, &range, &[], 100).expect("custom_events_report failed")
82+
});
83+
});
84+
group.finish();
85+
}
86+
7687
{
7788
let mut group = c.benchmark_group("dimension_report");
7889
configure_group(&mut group);

‎crates/liwan/src/app/core/events.rs‎

Lines changed: 60 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -360,11 +360,10 @@ fn update_event_times(conn: &Connection, from_time: DateTime<Utc>, entities: &[S
360360
}
361361

362362
let entity_vars = repeat_vars(entities.len());
363-
// this can probably be simplified, sadly the where clause can't contain window functions
364363
let sql = format!("--sql
365364
with
366365
filtered_events as (
367-
select *
366+
select rowid, entity_id, visitor_group_id, created_at
368367
from events
369368
where event = 'pageview' and entity_id in ({entity_vars}) and (created_at >= ?::timestamp or visitor_group_id in (
370369
select visitor_group_id
@@ -374,19 +373,17 @@ fn update_event_times(conn: &Connection, from_time: DateTime<Utc>, entities: &[S
374373
),
375374
cte as (
376375
select
377-
entity_id,
378-
visitor_group_id,
379-
created_at,
380-
created_at - lag(created_at) over (partition by entity_id, visitor_group_id order by created_at) as time_from_last_event,
381-
lead(created_at) over (partition by entity_id, visitor_group_id order by created_at) - created_at as time_to_next_event
376+
rowid,
377+
created_at - lag(created_at) over (partition by entity_id, visitor_group_id order by created_at, rowid) as time_from_last_event,
378+
lead(created_at) over (partition by entity_id, visitor_group_id order by created_at, rowid) - created_at as time_to_next_event
382379
from filtered_events
383380
)
384381
update events
385382
set
386383
time_from_last_event = cte.time_from_last_event,
387384
time_to_next_event = cte.time_to_next_event
388385
from cte
389-
where events.event = 'pageview' and events.entity_id = cte.entity_id and events.visitor_group_id = cte.visitor_group_id and events.created_at = cte.created_at;
386+
where events.rowid = cte.rowid;
390387
");
391388

392389
let mut params = ParamVec::new();
@@ -404,6 +401,61 @@ mod tests {
404401
use crate::app::Liwan;
405402
use crate::config::Config;
406403

404+
#[test]
405+
fn failed_append_does_not_update_cache() {
406+
let app = Liwan::new_memory(Config::default()).unwrap();
407+
let conn = app.events_conn().unwrap();
408+
conn.execute_batch("create unique index unique_visitor on events(visitor_group_id)").unwrap();
409+
let entities = ["entity-1".into()];
410+
assert!(!app.events.custom_data(&entities).unwrap().events);
411+
let mut first = event(Utc::now());
412+
first.event = "signup".into();
413+
let second = first.clone();
414+
415+
assert!(app.events.append([first, second].into_iter()).is_err());
416+
417+
assert_eq!(conn.query_row("select count(*) from events", [], |row| row.get::<_, u64>(0)).unwrap(), 0);
418+
assert!(!app.events.custom_data(&entities).unwrap().events);
419+
}
420+
421+
#[test]
422+
fn session_updates_identify_individual_rows() {
423+
let app = Liwan::new_memory(Config::default()).unwrap();
424+
let conn = app.events_conn().unwrap();
425+
conn.execute_batch(
426+
"insert into events (entity_id, visitor_group_id, event, created_at)
427+
values ('a', 'visitor', 'pageview', '2026-01-01 00:00:00'),
428+
('a', 'visitor', 'pageview', '2026-01-01 00:00:00'),
429+
('a', 'visitor', 'pageview', '2026-01-01 00:01:00'),
430+
('b', 'visitor', 'pageview', '2026-01-01 00:00:00'),
431+
('b', 'visitor', 'pageview', '2026-01-01 00:02:00'),
432+
('a', 'visitor', 'signup', '2026-01-01 00:00:30'),
433+
('c', 'visitor', 'pageview', '2026-01-01 00:00:00')",
434+
)
435+
.unwrap();
436+
let from = "2026-01-01T00:00:00Z".parse().unwrap();
437+
update_event_times(&conn, from, &["a".into(), "b".into()]).unwrap();
438+
let rows = conn
439+
.prepare("select epoch(time_from_last_event)::bigint, epoch(time_to_next_event)::bigint from events order by rowid")
440+
.unwrap()
441+
.query_map([], |row| Ok((row.get::<_, Option<i64>>(0)?, row.get::<_, Option<i64>>(1)?)))
442+
.unwrap()
443+
.collect::<DuckResult<Vec<_>>>()
444+
.unwrap();
445+
assert_eq!(
446+
rows,
447+
vec![
448+
(None, Some(0)),
449+
(Some(0), Some(60)),
450+
(Some(60), None),
451+
(None, Some(120)),
452+
(Some(120), None),
453+
(None, None),
454+
(None, None)
455+
]
456+
);
457+
}
458+
407459
fn event(created_at: DateTime<Utc>) -> Event {
408460
Event {
409461
entity_id: "entity-1".to_string(),

‎crates/liwan/src/app/core/reports/custom_events.rs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,16 @@ mod tests {
101101
assert_eq!(report.rows[0].completions, 3);
102102
assert_eq!(report.rows[0].uniques, 2);
103103

104+
let report = custom_events_report(&conn, &["a".into(), "b".into()], &range, &[], 10).unwrap();
105+
assert!(!report.truncated);
106+
assert_eq!(report.rows.len(), 2);
107+
assert_eq!(report.rows[0].completions, 4);
108+
assert_eq!(report.rows[0].uniques, 3);
109+
110+
let report = custom_events_report(&conn, &["a".into()], &range, &[], 0).unwrap();
111+
assert!(report.truncated);
112+
assert!(report.rows.is_empty());
113+
104114
let empty = DateRange { start: now - Duration::hours(3), end: now - Duration::hours(2) };
105115
assert!(custom_events_report(&conn, &["a".into()], &empty, &[], 10).unwrap().rows.is_empty());
106116

‎crates/liwan/src/app/core/reports/dimension.rs‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -66,9 +66,7 @@ pub fn dimension_report(
6666
Dimension::Orientation => ("orientation", None),
6767
Dimension::Property => match property_key {
6868
None => ("unnest(map_keys(properties))", Some("properties is not null".to_string())),
69-
Some(_) => {
70-
("map_extract_value(properties, ?)", Some("map_extract_value(properties, ?) is not null".to_string()))
71-
}
69+
Some(_) => ("map_extract_value(properties, ?)", Some("map_contains(properties, ?)".to_string())),
7270
},
7371
};
7472
let filters_sql = match (filters_sql.is_empty(), dimension_scope_sql) {
@@ -330,6 +328,7 @@ mod tests {
330328
assert_eq!(report("pageview", None, &[]), [row("plan", 3.0), row("ref", 1.0)]);
331329
assert_eq!(report("signup", None, &[]), [row("source", 1.0)]);
332330
assert_eq!(report("pageview", Some("plan"), &[]), [row("free", 1.0), row("pro", 2.0)]);
331+
assert!(report("pageview", Some("missing"), &[]).is_empty());
333332

334333
let filter = |key: &str, filter_type: FilterType, value: Option<&str>| DimensionFilter {
335334
dimension: Dimension::Property,

‎crates/liwan/src/app/core/reports/graph.rs‎

Lines changed: 41 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -198,25 +198,15 @@ pub fn overall_report(
198198
e.entity_id in ({entity_vars})
199199
{filters_sql}
200200
),
201-
bucketed_events as (
201+
event_bins as (
202202
select
203203
tb.bucket_idx,
204-
sd.visitor_group_id,
205-
sd.created_at,
206-
sd.time_from_last_event,
207-
sd.time_to_next_event,
208-
sd.exited_at
209-
from (select * from session_data order by created_at) sd
210-
asof join (select * from time_bins order by bin_start) tb
204+
{metric_sql} as metric_value
205+
from session_data sd
206+
asof join time_bins tb
211207
on sd.created_at >= tb.bin_start
212208
where sd.created_at < tb.bin_end
213-
),
214-
event_bins as (
215-
select
216-
sd.bucket_idx,
217-
{metric_sql} as metric_value
218-
from bucketed_events sd
219-
group by sd.bucket_idx
209+
group by tb.bucket_idx
220210
)
221211
select
222212
tb.bin_start,
@@ -392,6 +382,42 @@ mod tests {
392382
assert_eq!(values, vec![1.0, 1.0]);
393383
}
394384

385+
#[test]
386+
fn overall_report_handles_irregular_buckets() {
387+
let app = Liwan::new_memory(Config::default()).unwrap();
388+
let conn = app.events_conn().unwrap();
389+
let start = local_datetime(Tz::UTC, 2024, 1, 1, 0, 0);
390+
for (visitor, hours, previous, next, exit) in [
391+
("outside", -1, None, None, None),
392+
("one", 0, None, Some(60), None),
393+
("one", 1, Some(60), None, Some(120)),
394+
("gap", 4, None, None, None),
395+
("two", 5, None, None, Some(30)),
396+
("boundary", 6, None, None, None),
397+
] {
398+
conn.execute(
399+
"insert into events (entity_id, visitor_group_id, event, created_at, time_from_last_event, time_to_next_event, exited_at)
400+
values ('entity-1', ?, 'pageview', ?, ? * interval '1 second', ? * interval '1 second', ?)",
401+
duckdb::params![visitor, start + Duration::hours(hours), previous, next, exit.map(|seconds| start + Duration::hours(hours) + Duration::seconds(seconds))],
402+
).unwrap();
403+
}
404+
let range = DateRange { start: start - Duration::hours(2), end: start + Duration::hours(10) };
405+
let buckets: Vec<_> = [(0, 2), (5, 6), (7, 8)]
406+
.into_iter()
407+
.map(|(from, to)| DateRange { start: start + Duration::hours(from), end: start + Duration::hours(to) })
408+
.collect();
409+
for (metric, expected) in [
410+
(Metric::Views, vec![2.0, 1.0, 0.0]),
411+
(Metric::UniqueVisitors, vec![1.0, 1.0, 0.0]),
412+
(Metric::BounceRate, vec![0.0, 1.0, 0.0]),
413+
(Metric::AvgTimeOnSite, vec![90.0, 30.0, 0.0]),
414+
] {
415+
let report =
416+
overall_report(&conn, &["entity-1".into()], "pageview", &range, &buckets, &[], &metric).unwrap();
417+
assert_eq!(report.iter().map(|point| point.value).collect::<Vec<_>>(), expected);
418+
}
419+
}
420+
395421
#[test]
396422
fn build_graph_buckets_rejects_excessive_ranges() {
397423
let start = local_datetime(Tz::UTC, 2024, 1, 1, 0, 0);

0 commit comments

Comments
 (0)