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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
//! Does a JSON-column equality reach the row provider?
//!
//! Not a product module. `filterable_column_name` returns `None` for
//! `SchemaColumnType::Jsonb`, so a JSON-column equality cannot become an
//! `RowFilter`. What that costs is the open question: the predicate is
//! still *answered*, by DataFusion's `FilterExec` one layer above the
//! provider, so the rows are materialised and then discarded.
//!
//! The text spelling (`WHERE v = '{"n":1}'`) is a type error
//! (`LIX_ERROR_TYPE_MISMATCH`) and never runs, so it cannot be the shape under
//! test. This probe uses the spelling that type-checks,
//! `WHERE v = CAST('...' AS JSONB)`, and a String-column arm as the control: same
//! fixture size, same one-row answer, on a column that *is* pushable.
//!
//! Counted at the materialisation boundary (`apply_row_batch_filters`),
//! above its `filters.is_empty()` early return, because a profile of that
//! function reads identically for "nothing to filter" and "never ran".
use serde_json::json;
use crate::engine::Engine;
use crate::session::SessionContext;
use crate::storage_adapter::Memory;
async fn open_session() -> SessionContext<Memory> {
let storage = Memory::new();
Engine::initialize(storage.clone())
.await
.expect("engine should initialize");
let engine = Engine::new(storage.clone())
.await
.expect("engine should open");
engine.open_session().await.expect("session should open")
}
async fn register(session: &SessionContext<Memory>, key: &str) {
let schema = json!({
"$schema": "https://lix.dev/schema-v1.json",
"key": key,
"columns": [
{ "name": "id", "type": "text", "nullable": false },
{ "name": "k", "type": "text", "nullable": false },
{ "name": "v", "type": "jsonb", "nullable": false },
],
"primary_key": ["id"],
});
session
.execute(
"INSERT INTO lix_registered_schema (value) VALUES (CAST($1 AS JSONB))",
&[crate::Value::Text(schema.to_string())],
)
.await
.expect("schema should register");
}
async fn seed(session: &SessionContext<Memory>, table: &str, count: usize) {
const CHUNK: usize = 250;
let mut index = 0;
while index < count {
let end = (index + CHUNK).min(count);
let values = (index..end)
.map(|i| format!("('r-{i}', 'k-{i}', CAST('{{\"n\":{i}}}' AS JSONB))"))
.collect::<Vec<_>>()
.join(",");
session
.execute(
&format!("INSERT INTO {table} (id, k, v) VALUES {values}"),
&[],
)
.await
.expect("rows should insert");
index = end;
}
}
/// The two spellings, same fixture, same one-row answer.
///
/// Counted with `provider_rows_examined`, the engine's own instrument, whose
/// documentation states the exact contract this probe needs: it is recorded
/// *before* a provider applies its row filters, at **every route a row
/// surface can take**, so a route that never reaches
/// `apply_row_batch_filters` is still counted. A census at that one
/// function is blind to the columnar and overlay routes, which is how a
/// "zero frames" reading can mean "different route" rather than "did nothing".
/// It is also task-local rather than process-global, so it cannot bleed.
///
/// `scan_rows` is reported alongside it because the pair is what makes the
/// claim falsifiable: a change that only moves filtering earlier in the plan
/// moves `scan_rows` and leaves `provider_rows_examined` alone.
#[tokio::test]
#[ignore = "measurement probe, not a gate"]
async fn json_column_equality_materialization() {
let n: usize = std::env::var("LIX_JPP_ROWS")
.ok()
.and_then(|raw| raw.parse().ok())
.unwrap_or(10_000);
println!("jpp | arm,n,answer_rows,provider_rows_examined,scan_rows,examined_per_answer_row");
// Control: a String column, which IS pushable. Same fixture size, same
// one-row answer, and no indexed access path either - so it isolates the
// JSON refusal from "this column has no access path".
{
let session = open_session().await;
register(&session, "jppstr").await;
seed(&session, "jppstr", n).await;
let (result, profile) = session
.execute_profiled("SELECT id FROM jppstr WHERE k = 'k-7'", &[])
.await
.expect("string-column scan should run");
println!(
"string,{n},{},{},{},{}",
result.rows().len(),
profile.provider_rows_examined,
profile.scan_rows,
profile.provider_rows_examined as f64 / result.rows().len().max(1) as f64
);
assert_eq!(
result.rows().len(),
1,
"the string arm must answer exactly one row"
);
}
{
let session = open_session().await;
register(&session, "jppjson").await;
seed(&session, "jppjson", n).await;
// The text spelling is a type error and never runs. Asserted so this
// probe cannot silently measure the wrong shape.
let text_spelling = session
.execute("SELECT id FROM jppjson WHERE v = '{\"n\":7}'", &[])
.await;
println!(
"jpp | text_spelling_err={:?}",
text_spelling.as_ref().err().map(|error| error.code.clone())
);
assert!(
text_spelling.is_err(),
"the text spelling must remain a type error"
);
let (result, profile) = session
.execute_profiled(
"SELECT id FROM jppjson WHERE v = CAST('{\"n\":7}' AS JSONB)",
&[],
)
.await
.expect("json-column scan should run");
println!(
"json,{n},{},{},{},{}",
result.rows().len(),
profile.provider_rows_examined,
profile.scan_rows,
profile.provider_rows_examined as f64 / result.rows().len().max(1) as f64
);
assert_eq!(
result.rows().len(),
1,
"the json arm must answer exactly one row"
);
}
// A primary-key equality, which DOES have an exact access path. Present as
// the upper bound: it is what "the provider examined only what it needed"
// looks like on this fixture, so the two arms above can be read against a
// number the engine can actually reach rather than against zero.
{
let session = open_session().await;
register(&session, "jpppk").await;
seed(&session, "jpppk", n).await;
let (result, profile) = session
.execute_profiled("SELECT id FROM jpppk WHERE id = 'r-7'", &[])
.await
.expect("primary-key scan should run");
println!(
"primary_key,{n},{},{},{},{}",
result.rows().len(),
profile.provider_rows_examined,
profile.scan_rows,
profile.provider_rows_examined as f64 / result.rows().len().max(1) as f64
);
assert_eq!(result.rows().len(), 1, "the pk arm must answer one row");
}
}