1use duckdb::types::ValueRef;
2use duckdb::Connection;
3use std::path::Path;
4
5use crate::types::*;
6
7pub struct DuckdbBackend {
8 conn: Connection,
9 path: String,
10 parquet_table: Option<String>,
12}
13
14impl DuckdbBackend {
15 pub fn open(path: &str) -> Result<Self, String> {
16 let config = duckdb::Config::default()
17 .access_mode(duckdb::AccessMode::ReadOnly)
18 .map_err(|e| format!("Failed to configure DuckDB: {}", e))?;
19 let conn = Connection::open_with_flags(path, config)
20 .map_err(|e| format!("Failed to open DuckDB database: {}", e))?;
21 Ok(DuckdbBackend {
22 conn,
23 path: path.to_string(),
24 parquet_table: None,
25 })
26 }
27
28 pub fn open_parquet(path: &str) -> Result<Self, String> {
29 let conn = Connection::open_in_memory()
30 .map_err(|e| format!("Failed to open in-memory DuckDB: {}", e))?;
31
32 let table_name = Path::new(path)
33 .file_stem()
34 .and_then(|s| s.to_str())
35 .unwrap_or("data")
36 .to_string();
37
38 conn.execute_batch(&format!(
39 "CREATE VIEW \"{}\" AS SELECT * FROM read_parquet('{}')",
40 table_name,
41 path.replace('\'', "''")
42 ))
43 .map_err(|e| format!("Failed to read Parquet file: {}", e))?;
44
45 Ok(DuckdbBackend {
46 conn,
47 path: path.to_string(),
48 parquet_table: Some(table_name),
49 })
50 }
51
52 fn is_parquet(&self) -> bool {
53 self.parquet_table.is_some()
54 }
55
56 pub fn path(&self) -> &str {
57 &self.path
58 }
59
60 pub fn get_info(&self) -> Result<DbInfo, String> {
61 let version: String = self
62 .conn
63 .query_row("SELECT library_version FROM pragma_version()", [], |row| {
64 row.get(0)
65 })
66 .map_err(|e| e.to_string())?;
67
68 let file_size = Path::new(&self.path)
69 .metadata()
70 .map(|m| m.len())
71 .unwrap_or(0);
72
73 let tables = self.list_tables()?;
74
75 let engine = if self.is_parquet() {
76 "Parquet"
77 } else {
78 "DuckDB"
79 };
80
81 Ok(DbInfo {
82 path: self.path.clone(),
83 file_size,
84 engine: engine.to_string(),
85 engine_version: version,
86 page_count: None,
87 page_size: None,
88 table_count: tables.len(),
89 })
90 }
91
92 pub fn list_tables(&self) -> Result<Vec<TableInfo>, String> {
93 if let Some(table_name) = &self.parquet_table {
94 let row_count = self.get_row_count(table_name).unwrap_or(0);
95 let columns = self.get_schema(table_name).unwrap_or_default();
96 return Ok(vec![TableInfo {
97 name: table_name.clone(),
98 row_count,
99 column_count: columns.len(),
100 }]);
101 }
102
103 let mut stmt = self
104 .conn
105 .prepare(
106 "SELECT table_name FROM information_schema.tables \
107 WHERE table_schema = 'main' AND table_type = 'BASE TABLE' \
108 ORDER BY table_name",
109 )
110 .map_err(|e| e.to_string())?;
111
112 let names: Vec<String> = stmt
113 .query_map([], |row| row.get(0))
114 .map_err(|e| e.to_string())?
115 .filter_map(|r| r.ok())
116 .collect();
117
118 let mut tables = Vec::new();
119 for name in names {
120 let row_count = self.get_row_count(&name).unwrap_or(0);
121 let columns = self.get_schema(&name).unwrap_or_default();
122 tables.push(TableInfo {
123 name,
124 row_count,
125 column_count: columns.len(),
126 });
127 }
128 Ok(tables)
129 }
130
131 pub fn list_views(&self) -> Result<Vec<String>, String> {
132 if self.is_parquet() {
133 return Ok(Vec::new());
134 }
135
136 let mut stmt = self
137 .conn
138 .prepare(
139 "SELECT table_name FROM information_schema.tables \
140 WHERE table_schema = 'main' AND table_type = 'VIEW' \
141 ORDER BY table_name",
142 )
143 .map_err(|e| e.to_string())?;
144
145 let views = stmt
146 .query_map([], |row| row.get(0))
147 .map_err(|e| e.to_string())?
148 .filter_map(|r| r.ok())
149 .collect();
150
151 Ok(views)
152 }
153
154 pub fn list_indexes(&self) -> Result<Vec<IndexInfo>, String> {
155 if self.is_parquet() {
156 return Ok(Vec::new());
157 }
158
159 let mut stmt = self
160 .conn
161 .prepare(
162 "SELECT index_name, table_name, is_unique \
163 FROM duckdb_indexes() \
164 WHERE schema_name = 'main' \
165 ORDER BY index_name",
166 )
167 .map_err(|e| e.to_string())?;
168
169 let indexes: Vec<IndexInfo> = stmt
170 .query_map([], |row| {
171 Ok(IndexInfo {
172 name: row.get(0)?,
173 table_name: row.get(1)?,
174 unique: row.get(2)?,
175 columns: Vec::new(),
176 })
177 })
178 .map_err(|e| e.to_string())?
179 .filter_map(|r| r.ok())
180 .collect();
181
182 Ok(indexes)
183 }
184
185 pub fn get_schema(&self, table: &str) -> Result<Vec<ColumnInfo>, String> {
186 let mut stmt = self
187 .conn
188 .prepare(
189 "SELECT column_name, data_type, is_nullable, column_default \
190 FROM information_schema.columns \
191 WHERE table_schema = 'main' AND table_name = ? \
192 ORDER BY ordinal_position",
193 )
194 .map_err(|e| e.to_string())?;
195
196 let columns = stmt
197 .query_map([table], |row| {
198 let nullable_str: String = row.get(2)?;
199 Ok(ColumnInfo {
200 name: row.get(0)?,
201 col_type: row.get(1)?,
202 nullable: nullable_str == "YES",
203 primary_key: false,
204 default_value: row.get(3).ok(),
205 })
206 })
207 .map_err(|e| e.to_string())?
208 .filter_map(|r| r.ok())
209 .collect();
210
211 Ok(columns)
212 }
213
214 pub fn query_table(
215 &self,
216 table: &str,
217 limit: usize,
218 offset: usize,
219 sort: Option<Sort>,
220 ) -> Result<QueryResult, String> {
221 let sql = format!(
222 "SELECT * FROM \"{}\"{} LIMIT {} OFFSET {}",
223 table, order_clause(&sort), limit, offset
224 );
225
226 let total = self.get_row_count(table).ok();
227 let mut result = self.run_query(&sql)?;
228 result.total_rows = total;
229 Ok(result)
230 }
231
232 pub fn run_query(&self, sql: &str) -> Result<QueryResult, String> {
233 let mut stmt = self.conn.prepare(sql).map_err(|e| e.to_string())?;
234 let mut result_rows = stmt.query([]).map_err(|e| e.to_string())?;
235
236 let columns: Vec<String> = result_rows
237 .as_ref()
238 .expect("query should return rows")
239 .column_names();
240
241 let col_count = columns.len();
242 let mut rows = Vec::new();
243
244 while let Some(row) = result_rows.next().map_err(|e| e.to_string())? {
245 let mut cells = Vec::new();
246 for i in 0..col_count {
247 let val = match row.get_ref(i) {
248 Ok(ValueRef::Null) => CellValue::Null,
249 Ok(ValueRef::Int(n)) => CellValue::Integer(n as i64),
250 Ok(ValueRef::BigInt(n)) => CellValue::Integer(n),
251 Ok(ValueRef::TinyInt(n)) => CellValue::Integer(n as i64),
252 Ok(ValueRef::SmallInt(n)) => CellValue::Integer(n as i64),
253 Ok(ValueRef::HugeInt(n)) => CellValue::Text(n.to_string()),
254 Ok(ValueRef::Float(f)) => CellValue::Real(f as f64),
255 Ok(ValueRef::Double(f)) => CellValue::Real(f),
256 Ok(ValueRef::Text(s)) => {
257 CellValue::Text(String::from_utf8_lossy(s).to_string())
258 }
259 Ok(ValueRef::Blob(b)) => CellValue::Blob(b.to_vec()),
260 Ok(other) => CellValue::Text(format!("{:?}", other)),
261 Err(_) => CellValue::Null,
262 };
263 cells.push(val);
264 }
265 rows.push(cells);
266 }
267
268 Ok(QueryResult {
269 columns,
270 rows,
271 total_rows: None,
272 })
273 }
274
275 pub fn get_row_count(&self, table: &str) -> Result<u64, String> {
276 self.conn
277 .query_row(
278 &format!("SELECT COUNT(*) FROM \"{}\"", table),
279 [],
280 |row| row.get::<_, i64>(0),
281 )
282 .map(|n| n as u64)
283 .map_err(|e| e.to_string())
284 }
285
286}