1use akar_extension::{Extension, ExtensionContext};
11use std::sync::Arc;
12
13#[cfg(feature = "bundled")]
18fn sqlite_value_to_string(val: &rusqlite::types::Value) -> String {
19 match val {
20 rusqlite::types::Value::Null => "NULL".into(),
21 rusqlite::types::Value::Integer(i) => i.to_string(),
22 rusqlite::types::Value::Real(f) => f.to_string(),
23 rusqlite::types::Value::Text(s) => s.clone(),
24 rusqlite::types::Value::Blob(b) => format!("<blob:{} bytes>", b.len()),
25 }
26}
27
28pub struct SqliteExtension;
30
31impl Default for SqliteExtension {
32 fn default() -> Self {
33 Self::new()
34 }
35}
36
37impl SqliteExtension {
38 pub fn new() -> Self {
39 Self
40 }
41}
42
43impl Extension for SqliteExtension {
44 fn name(&self) -> &'static str {
45 "SQLITE"
46 }
47
48 fn load(&self, context: &ExtensionContext) -> Result<(), String> {
49 #[allow(unused_imports)]
50 use akar_function::Value;
51 use akar_function::registry::{ScalarFunction, TableFunction};
52
53 #[cfg(feature = "bundled")]
55 {
56 let query_fn: Arc<dyn Fn(&[Value]) -> Result<Value, String> + Send + Sync> = Arc::new(|args| {
57 if args.len() < 2 {
58 return Err("sqlite_query requires (path, sql) arguments".into());
59 }
60 let path = match &args[0] {
61 Value::String(s) => s.clone(),
62 _ => return Err("sqlite_query: first argument must be a path string".into()),
63 };
64 let sql = match &args[1] {
65 Value::String(s) => s.clone(),
66 _ => return Err("sqlite_query: second argument must be a SQL string".into()),
67 };
68
69 let conn =
70 rusqlite::Connection::open(&path).map_err(|e| format!("Failed to open SQLite DB '{path}': {e}"))?;
71
72 let mut stmt = conn.prepare(&sql).map_err(|e| format!("SQLite prepare error: {e}"))?;
73
74 let col_count = stmt.column_count();
75 let mut rows = stmt.query([]).map_err(|e| format!("SQLite query error: {e}"))?;
76
77 let mut parts = Vec::new();
79 while let Some(row) = rows.next().map_err(|e| format!("SQLite row error: {e}"))? {
80 for i in 0..col_count {
81 let val: rusqlite::types::Value = row
82 .get::<_, rusqlite::types::Value>(i)
83 .unwrap_or(rusqlite::types::Value::Null);
84 parts.push(sqlite_value_to_string(&val));
85 }
86 }
87 if parts.is_empty() {
88 Ok(Value::String("(empty)".into()))
89 } else {
90 Ok(Value::String(parts.join(",")))
91 }
92 });
93
94 context.register_scalar_function(
95 "sqlite_query",
96 ScalarFunction::CustomScalar {
97 name: "sqlite_query".into(),
98 execute: query_fn,
99 },
100 );
101
102 let scan_fn: Arc<dyn Fn(&[Value], &mut akar_function::DataChunk) -> Result<(), String> + Send + Sync> =
104 Arc::new(|args, chunk| {
105 if args.len() < 2 {
106 return Err("sqlite_scan requires (path, table) arguments".into());
107 }
108 let path = match &args[0] {
109 Value::String(s) => s.clone(),
110 _ => return Err("sqlite_scan: first argument must be a path string".into()),
111 };
112 let table = match &args[1] {
113 Value::String(s) => s.clone(),
114 _ => return Err("sqlite_scan: second argument must be a table name".into()),
115 };
116
117 if chunk.size > 0 {
118 return Ok(());
119 }
120
121 let conn = rusqlite::Connection::open(&path)
122 .map_err(|e| format!("Failed to open SQLite DB '{path}': {e}"))?;
123
124 let sql = format!(
125 "SELECT * FROM {}",
126 akar_common::extension_utils::quote_sql_table_name(&table)
127 );
128 let mut stmt = conn.prepare(&sql).map_err(|e| format!("SQLite prepare error: {e}"))?;
129 let col_count = stmt.column_count();
130 let names: Vec<String> = (0..col_count)
131 .map(|i| stmt.column_name(i).map(str::to_string).unwrap_or_default())
132 .collect();
133 let mut rows = stmt.query([]).map_err(|e| format!("SQLite query error: {e}"))?;
134
135 let mut columns: Vec<Vec<Option<String>>> = vec![Vec::new(); col_count];
136 while let Some(row) = rows.next().map_err(|e| format!("SQLite row error: {e}"))? {
137 for i in 0..col_count {
138 let val: rusqlite::types::Value = row
139 .get::<_, rusqlite::types::Value>(i)
140 .unwrap_or(rusqlite::types::Value::Null);
141 columns[i].push(Some(sqlite_value_to_string(&val)));
142 }
143 }
144
145 chunk.fields.clear();
146 chunk.field_types.clear();
147 chunk.field_names.clear();
148 for (col, name) in columns.into_iter().zip(names) {
149 chunk
150 .fields
151 .push(std::sync::Arc::new(arrow::array::StringArray::from_iter(
152 col.iter().map(|o| o.as_deref()),
153 )) as arrow::array::ArrayRef);
154 chunk.field_types.push(akar_common::types::PhysicalTypeID::String);
155 chunk.field_names.push(name);
156 }
157 chunk.size = chunk.fields.first().map(|f| f.len()).unwrap_or(0);
158 Ok(())
159 });
160
161 context.register_table_function(
162 "sqlite_scan",
163 TableFunction::CustomTable {
164 name: "sqlite_scan".into(),
165 execute: scan_fn,
166 },
167 );
168
169 tracing::info!("SQLite extension loaded: 2 functions registered (rusqlite native)");
170 }
171
172 #[cfg(not(feature = "bundled"))]
173 {
174 context.register_scalar_function(
175 "sqlite_query",
176 ScalarFunction::CustomScalar {
177 name: "sqlite_query".into(),
178 execute: Arc::new(|_| Err("SQLite not available (feature 'bundled' disabled)".into())),
179 },
180 );
181 context.register_table_function(
182 "sqlite_scan",
183 TableFunction::CustomTable {
184 name: "sqlite_scan".into(),
185 execute: Arc::new(|_, _| Err("SQLite not available (feature 'bundled' disabled)".into())),
186 },
187 );
188 tracing::info!("SQLite extension loaded: 2 functions registered (placeholder)");
189 }
190
191 Ok(())
192 }
193}
194
195#[cfg(test)]
196mod tests {
197 use super::*;
198
199 #[test]
200 fn test_sqlite_extension_name() {
201 let ext = SqliteExtension::new();
202 assert_eq!(ext.name(), "SQLITE");
203 }
204}