1use duckdb::types::{TimeUnit, 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(v) => cell_from(v),
249 Err(_) => CellValue::Null,
250 };
251 cells.push(val);
252 }
253 rows.push(cells);
254 }
255
256 Ok(QueryResult {
257 columns,
258 rows,
259 total_rows: None,
260 })
261 }
262
263 pub fn get_row_count(&self, table: &str) -> Result<u64, String> {
264 self.conn
265 .query_row(
266 &format!("SELECT COUNT(*) FROM \"{}\"", table),
267 [],
268 |row| row.get::<_, i64>(0),
269 )
270 .map(|n| n as u64)
271 .map_err(|e| e.to_string())
272 }
273
274}
275
276fn cell_from(value: ValueRef<'_>) -> CellValue {
283 match value {
284 ValueRef::Null => CellValue::Null,
285 ValueRef::Boolean(b) => CellValue::Text(b.to_string()),
286 ValueRef::TinyInt(n) => CellValue::Integer(n as i64),
287 ValueRef::SmallInt(n) => CellValue::Integer(n as i64),
288 ValueRef::Int(n) => CellValue::Integer(n as i64),
289 ValueRef::BigInt(n) => CellValue::Integer(n),
290 ValueRef::UTinyInt(n) => CellValue::Integer(n as i64),
291 ValueRef::USmallInt(n) => CellValue::Integer(n as i64),
292 ValueRef::UInt(n) => CellValue::Integer(n as i64),
293 ValueRef::UBigInt(n) => match i64::try_from(n) {
296 Ok(v) => CellValue::Integer(v),
297 Err(_) => CellValue::Text(n.to_string()),
298 },
299 ValueRef::HugeInt(n) => CellValue::Text(n.to_string()),
300 ValueRef::UHugeInt(n) => CellValue::Text(n.to_string()),
301 ValueRef::Float(f) => CellValue::Real(f as f64),
302 ValueRef::Double(f) => CellValue::Real(f),
303 ValueRef::Decimal(d) => CellValue::Text(d.to_string()),
304 ValueRef::Date32(days) => CellValue::Text(format_date(days)),
305 ValueRef::Time64(unit, v) => CellValue::Text(format_time(unit, v)),
306 ValueRef::Timestamp(unit, v) => CellValue::Text(format_timestamp(unit, v)),
307 ValueRef::Interval { months, days, nanos } => {
308 CellValue::Text(format_interval(months, days, nanos))
309 }
310 ValueRef::Text(s) => CellValue::Text(String::from_utf8_lossy(s).to_string()),
311 ValueRef::Blob(b) | ValueRef::Geometry(b) => CellValue::Blob(b.to_vec()),
312 other => CellValue::Text(format!("{:?}", other)),
313 }
314}
315
316fn civil_from_days(days: i32) -> (i32, u32, u32) {
320 let z = days as i64 + 719_468;
321 let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
322 let doe = z - era * 146_097;
323 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
324 let y = yoe + era * 400;
325 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
326 let mp = (5 * doy + 2) / 153;
327 let d = doy - (153 * mp + 2) / 5 + 1;
328 let m = if mp < 10 { mp + 3 } else { mp - 9 };
329 ((if m <= 2 { y + 1 } else { y }) as i32, m as u32, d as u32)
330}
331
332fn format_date(days: i32) -> String {
333 let (y, m, d) = civil_from_days(days);
334 format!("{:04}-{:02}-{:02}", y, m, d)
335}
336
337fn split_seconds(unit: TimeUnit, value: i64) -> (i64, String) {
340 let (per_second, digits) = match unit {
341 TimeUnit::Second => (1i64, 0usize),
342 TimeUnit::Millisecond => (1_000, 3),
343 TimeUnit::Microsecond => (1_000_000, 6),
344 TimeUnit::Nanosecond => (1_000_000_000, 9),
345 };
346 let seconds = value.div_euclid(per_second);
348 let frac = value.rem_euclid(per_second);
349 if digits == 0 || frac == 0 {
350 return (seconds, String::new());
351 }
352 let text = format!("{:0width$}", frac, width = digits);
353 (seconds, text.trim_end_matches('0').to_string())
354}
355
356fn format_clock(seconds_of_day: i64, frac: &str) -> String {
357 let (h, m, s) = (
358 seconds_of_day / 3600,
359 (seconds_of_day / 60) % 60,
360 seconds_of_day % 60,
361 );
362 if frac.is_empty() {
363 format!("{:02}:{:02}:{:02}", h, m, s)
364 } else {
365 format!("{:02}:{:02}:{:02}.{}", h, m, s, frac)
366 }
367}
368
369fn format_time(unit: TimeUnit, value: i64) -> String {
370 let (seconds, frac) = split_seconds(unit, value);
371 format_clock(seconds.rem_euclid(86_400), &frac)
372}
373
374fn format_timestamp(unit: TimeUnit, value: i64) -> String {
375 let (seconds, frac) = split_seconds(unit, value);
376 let date = format_date(seconds.div_euclid(86_400) as i32);
377 format!("{} {}", date, format_clock(seconds.rem_euclid(86_400), &frac))
378}
379
380fn format_interval(months: i32, days: i32, nanos: i64) -> String {
383 let mut parts = Vec::new();
384 if months != 0 {
385 let (years, rem) = (months / 12, months % 12);
386 if years != 0 {
387 parts.push(format!("{} year{}", years, if years.abs() == 1 { "" } else { "s" }));
388 }
389 if rem != 0 {
390 parts.push(format!("{} month{}", rem, if rem.abs() == 1 { "" } else { "s" }));
391 }
392 }
393 if days != 0 {
394 parts.push(format!("{} day{}", days, if days.abs() == 1 { "" } else { "s" }));
395 }
396 if nanos != 0 || parts.is_empty() {
397 let (seconds, frac) = split_seconds(TimeUnit::Nanosecond, nanos);
398 parts.push(format_clock(seconds, &frac));
399 }
400 parts.join(" ")
401}
402
403#[cfg(test)]
404mod format_tests {
405 use super::*;
406
407 #[test]
408 fn dates_round_trip_known_days() {
409 assert_eq!(format_date(0), "1970-01-01");
410 assert_eq!(format_date(19737), "2024-01-15");
411 assert_eq!(format_date(19781), "2024-02-28");
413 assert_eq!(format_date(19782), "2024-02-29");
414 assert_eq!(format_date(19783), "2024-03-01");
415 assert_eq!(format_date(19051), "2022-02-28");
417 assert_eq!(format_date(19052), "2022-03-01");
418 assert_eq!(format_date(-1), "1969-12-31");
420 assert_eq!(format_date(-719_162), "0001-01-01");
421 }
422
423 #[test]
424 fn times_drop_a_zero_fraction_and_keep_a_real_one() {
425 assert_eq!(format_time(TimeUnit::Microsecond, 49_530_000_000), "13:45:30");
426 assert_eq!(format_time(TimeUnit::Microsecond, 49_530_000_500), "13:45:30.0005");
427 assert_eq!(format_time(TimeUnit::Second, 0), "00:00:00");
428 assert_eq!(format_time(TimeUnit::Millisecond, 86_399_999), "23:59:59.999");
429 }
430
431 #[test]
432 fn timestamps_carry_both_halves() {
433 assert_eq!(
434 format_timestamp(TimeUnit::Microsecond, 1_705_326_330_000_000),
435 "2024-01-15 13:45:30"
436 );
437 assert_eq!(format_timestamp(TimeUnit::Second, 0), "1970-01-01 00:00:00");
438 assert_eq!(format_timestamp(TimeUnit::Second, -1), "1969-12-31 23:59:59");
440 }
441
442 #[test]
443 fn intervals_report_each_field_as_given() {
444 assert_eq!(format_interval(14, 3, 3_600_000_000_000), "1 year 2 months 3 days 01:00:00");
445 assert_eq!(format_interval(1, 0, 0), "1 month");
446 assert_eq!(format_interval(0, 1, 0), "1 day");
447 assert_eq!(format_interval(0, 0, 0), "00:00:00");
449 }
450}