akar-processor 0.2.1

Query processor and execution engine for the Akar embedded graph database
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

//! PhysicalScan throughput benchmarks at various table sizes.

use akar_common::enums::CompressionType;
use akar_common::types::{LogicalTypeID, Value};
use akar_processor::physical_operator::{PhysicalOperatorExec, PhysicalScan};
use akar_storage::table::ColumnDefinition;
use criterion::{Criterion, criterion_group, criterion_main};
use std::hint::black_box;

/// Generate a Vec<Vec<Value>> table with num_rows rows and the given schema.
fn generate_table(num_rows: usize, columns: &[ColumnDefinition]) -> Vec<Vec<Value>> {
    let num_cols = columns.len();
    let mut data: Vec<Vec<Value>> = (0..num_cols).map(|_| Vec::with_capacity(num_rows)).collect();

    for row in 0..num_rows {
        for (col, def) in columns.iter().enumerate() {
            let val = match def.logical_type {
                LogicalTypeID::Int64 => Value::Int64(row as i64),
                LogicalTypeID::Double => Value::Double(row as f64 * 1.5),
                LogicalTypeID::Bool => Value::Bool(row % 2 == 0),
                LogicalTypeID::String => Value::String(format!("val_{row}")),
                _ => Value::Int64(row as i64),
            };
            data[col].push(val);
        }
    }
    data
}

fn make_scan(table_data: Vec<Vec<Value>>, columns: Vec<ColumnDefinition>) -> PhysicalScan {
    let num_rows = table_data.first().map(|c| c.len()).unwrap_or(0);
    let mut scan = PhysicalScan::new("BenchTable".into(), 0, num_rows.max(1) as u64);
    scan = scan.with_data(table_data, columns);
    scan
}

fn schema() -> Vec<ColumnDefinition> {
    vec![
        ColumnDefinition {
            name: String::from("id"),
            logical_type: LogicalTypeID::Int64,
            is_primary_key: true,
            compression: CompressionType::Uncompressed,
        },
        ColumnDefinition {
            name: String::from("name"),
            logical_type: LogicalTypeID::String,
            is_primary_key: false,
            compression: CompressionType::Uncompressed,
        },
        ColumnDefinition {
            name: String::from("score"),
            logical_type: LogicalTypeID::Double,
            is_primary_key: false,
            compression: CompressionType::Uncompressed,
        },
        ColumnDefinition {
            name: String::from("active"),
            logical_type: LogicalTypeID::Bool,
            is_primary_key: false,
            compression: CompressionType::Uncompressed,
        },
    ]
}

fn bench_scan_100(c: &mut Criterion) {
    let columns = schema();
    let data = generate_table(100, &columns);
    let scan = make_scan(data, columns);
    c.bench_function("scan/100_rows", |b| {
        b.iter(|| {
            let result = scan.execute(black_box(vec![]));
            black_box(result.unwrap());
        })
    });
}

fn bench_scan_1k(c: &mut Criterion) {
    let columns = schema();
    let data = generate_table(1_000, &columns);
    let scan = make_scan(data, columns);
    c.bench_function("scan/1k_rows", |b| {
        b.iter(|| {
            let result = scan.execute(black_box(vec![]));
            black_box(result.unwrap());
        })
    });
}

fn bench_scan_10k(c: &mut Criterion) {
    let columns = schema();
    let data = generate_table(10_000, &columns);
    let scan = make_scan(data, columns);
    c.bench_function("scan/10k_rows", |b| {
        b.iter(|| {
            let result = scan.execute(black_box(vec![]));
            black_box(result.unwrap());
        })
    });
}

fn bench_scan_selective_columns(c: &mut Criterion) {
    let columns = schema();
    let data = generate_table(10_000, &columns);
    let mut scan = make_scan(data, columns);
    scan = scan.with_columns(vec![0, 2]); // only id + score
    c.bench_function("scan/10k_selective_2_of_4_cols", |b| {
        b.iter(|| {
            let result = scan.execute(black_box(vec![]));
            black_box(result.unwrap());
        })
    });
}

criterion_group!(
    benches,
    bench_scan_100,
    bench_scan_1k,
    bench_scan_10k,
    bench_scan_selective_columns,
);
criterion_main!(benches);