opendal_service_sqlite/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::str::FromStr;
19use std::sync::Arc;
20
21use asyncband::once::OnceCell;
22use sqlx::sqlite::SqliteConnectOptions;
23
24use super::SQLITE_SCHEME;
25use super::config::SqliteConfig;
26use super::core::SqliteCore;
27use super::deleter::SqliteDeleter;
28use super::reader::*;
29use super::writer::SqliteWriter;
30use opendal_core::raw::oio;
31use opendal_core::raw::*;
32use opendal_core::*;
33
34#[doc = include_str!("docs.md")]
35#[derive(Debug, Default)]
36pub struct SqliteBuilder {
37    pub(super) config: SqliteConfig,
38}
39
40impl SqliteBuilder {
41    /// Set the connection_string of the sqlite service.
42    ///
43    /// This connection string is used to connect to the sqlite service. There are url based formats:
44    ///
45    /// ## Url
46    ///
47    /// This format resembles the url format of the sqlite client:
48    ///
49    /// - `sqlite::memory:`
50    /// - `sqlite:data.db`
51    /// - `sqlite://data.db`
52    ///
53    /// For more information, please visit <https://docs.rs/sqlx/latest/sqlx/sqlite/struct.SqliteConnectOptions.html>.
54    pub fn connection_string(mut self, v: &str) -> Self {
55        if !v.is_empty() {
56            self.config.connection_string = Some(v.to_string());
57        }
58        self
59    }
60
61    /// set the working directory, all operations will be performed under it.
62    ///
63    /// default: "/"
64    pub fn root(mut self, root: &str) -> Self {
65        self.config.root = if root.is_empty() {
66            None
67        } else {
68            Some(root.to_string())
69        };
70
71        self
72    }
73
74    /// Set the table name of the sqlite service to read/write.
75    pub fn table(mut self, table: &str) -> Self {
76        if !table.is_empty() {
77            self.config.table = Some(table.to_string());
78        }
79        self
80    }
81
82    /// Set the key field name of the sqlite service to read/write.
83    ///
84    /// Default to `key` if not specified.
85    pub fn key_field(mut self, key_field: &str) -> Self {
86        if !key_field.is_empty() {
87            self.config.key_field = Some(key_field.to_string());
88        }
89        self
90    }
91
92    /// Set the value field name of the sqlite service to read/write.
93    ///
94    /// Default to `value` if not specified.
95    pub fn value_field(mut self, value_field: &str) -> Self {
96        if !value_field.is_empty() {
97            self.config.value_field = Some(value_field.to_string());
98        }
99        self
100    }
101}
102
103impl Builder for SqliteBuilder {
104    type Config = SqliteConfig;
105
106    fn build(self) -> Result<impl Service> {
107        let conn = match self.config.connection_string {
108            Some(v) => v,
109            None => {
110                return Err(Error::new(
111                    ErrorKind::ConfigInvalid,
112                    "connection_string is required but not set",
113                )
114                .with_context("service", SQLITE_SCHEME));
115            }
116        };
117
118        let config = SqliteConnectOptions::from_str(&conn).map_err(|err| {
119            Error::new(ErrorKind::ConfigInvalid, "connection_string is invalid")
120                .with_context("service", SQLITE_SCHEME)
121                .set_source(err)
122        })?;
123
124        let table = match self.config.table {
125            Some(v) => v,
126            None => {
127                return Err(Error::new(ErrorKind::ConfigInvalid, "table is empty")
128                    .with_context("service", SQLITE_SCHEME));
129            }
130        };
131
132        let key_field = self.config.key_field.unwrap_or_else(|| "key".to_string());
133
134        let value_field = self
135            .config
136            .value_field
137            .unwrap_or_else(|| "value".to_string());
138
139        let root = normalize_root(self.config.root.as_deref().unwrap_or("/"));
140
141        Ok(SqliteBackend::new(SqliteCore {
142            pool: OnceCell::new(),
143            config,
144            table,
145            key_field,
146            value_field,
147        })
148        .with_normalized_root(root))
149    }
150}
151
152pub fn parse_sqlite_error(err: sqlx::Error) -> Error {
153    let is_temporary = matches!(
154        &err,
155        sqlx::Error::Database(db_err) if db_err.code().is_some_and(|c| c == "5" || c == "6")
156    );
157
158    let message = if is_temporary {
159        "database is locked or busy"
160    } else {
161        "unhandled error from sqlite"
162    };
163
164    let mut error = Error::new(ErrorKind::Unexpected, message).set_source(err);
165    if is_temporary {
166        error = error.set_temporary();
167    }
168    error
169}
170
171/// SqliteBackend implements [`Service`] for SQLite-backed object storage.
172#[derive(Debug, Clone)]
173pub struct SqliteBackend {
174    pub(crate) core: Arc<SqliteCore>,
175    pub(crate) root: String,
176    pub(crate) info: ServiceInfo,
177    pub(crate) capability: Capability,
178}
179
180impl SqliteBackend {
181    fn new(core: SqliteCore) -> Self {
182        let info = ServiceInfo::new(SQLITE_SCHEME, "/", &core.table);
183        let capability = Capability {
184            read: true,
185            write: true,
186            create_dir: true,
187            delete: true,
188            stat: true,
189            write_can_empty: true,
190            list: false,
191            ..Default::default()
192        };
193
194        Self {
195            core: Arc::new(core),
196            root: "/".to_string(),
197            info,
198            capability,
199        }
200    }
201
202    fn with_normalized_root(mut self, root: String) -> Self {
203        self.info = self.info.with_root(&root);
204        self.root = root;
205        self
206    }
207}
208
209impl Service for SqliteBackend {
210    type Reader = oio::StreamReader<SqliteReader>;
211    type Writer = SqliteWriter;
212    type Lister = ();
213    type Deleter = oio::OneShotDeleter<SqliteDeleter>;
214    type Copier = ();
215    type Composer = ();
216
217    fn info(&self) -> ServiceInfo {
218        self.info.clone()
219    }
220
221    fn capability(&self) -> Capability {
222        self.capability
223    }
224
225    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
226        let p = build_abs_path(&self.root, path);
227
228        if p == build_abs_path(&self.root, "") {
229            Ok(RpStat::new(MetadataBuilder::dir().build()))
230        } else {
231            let length = self.core.get_length(&p).await?;
232            match length {
233                Some(length) => {
234                    let metadata = if p.ends_with('/') {
235                        MetadataBuilder::dir()
236                    } else {
237                        MetadataBuilder::file(length as u64)
238                    };
239                    Ok(RpStat::new(metadata.build()))
240                }
241                None => {
242                    // Check if this might be a directory by looking for keys with this prefix
243                    let dir_path = if p.ends_with('/') {
244                        p.clone()
245                    } else {
246                        format!("{}/", p)
247                    };
248                    let count = self.core.count_under(&dir_path).await?;
249
250                    if count > 0 {
251                        // Directory exists (has children)
252                        Ok(RpStat::new(MetadataBuilder::dir().build()))
253                    } else {
254                        Err(Error::new(ErrorKind::NotFound, "key not found in sqlite"))
255                    }
256                }
257            }
258        }
259    }
260    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
261        let output: oio::StreamReader<SqliteReader> = {
262            Ok(oio::StreamReader::new(SqliteReader::new(
263                self.clone(),
264                path,
265                args,
266            )))
267        }?;
268
269        Ok(output)
270    }
271
272    fn write(&self, _ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
273        let output: SqliteWriter = {
274            let p = build_abs_path(&self.root, path);
275            Ok(SqliteWriter::new(self.core.clone(), &p))
276        }?;
277
278        Ok(output)
279    }
280
281    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
282        let output: oio::OneShotDeleter<SqliteDeleter> = {
283            Ok(oio::OneShotDeleter::new(SqliteDeleter::new(
284                self.core.clone(),
285                self.root.clone(),
286            )))
287        }?;
288
289        Ok(output)
290    }
291
292    async fn create_dir(
293        &self,
294        _ctx: &OperationContext,
295        path: &str,
296        _: OpCreateDir,
297    ) -> Result<RpCreateDir> {
298        let p = build_abs_path(&self.root, path);
299
300        // Ensure path ends with '/' for directory marker
301        let dir_path = if p.ends_with('/') {
302            p
303        } else {
304            format!("{}/", p)
305        };
306
307        // Store directory marker with empty content
308        self.core.set(&dir_path, Buffer::new()).await?;
309
310        Ok(RpCreateDir::default())
311    }
312
313    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
314        Err(Error::new(
315            ErrorKind::Unsupported,
316            "operation is not supported",
317        ))
318    }
319
320    fn copy(
321        &self,
322        _ctx: &OperationContext,
323        _from: &str,
324        _to: &str,
325        _args: OpCopy,
326    ) -> Result<Self::Copier> {
327        Err(Error::new(
328            ErrorKind::Unsupported,
329            "operation is not supported",
330        ))
331    }
332
333    async fn rename(
334        &self,
335        _ctx: &OperationContext,
336        _from: &str,
337        _to: &str,
338        _args: OpRename,
339    ) -> Result<RpRename> {
340        Err(Error::new(
341            ErrorKind::Unsupported,
342            "operation is not supported",
343        ))
344    }
345
346    async fn presign(
347        &self,
348        _ctx: &OperationContext,
349        _path: &str,
350        _args: OpPresign,
351    ) -> Result<RpPresign> {
352        Err(Error::new(
353            ErrorKind::Unsupported,
354            "operation is not supported",
355        ))
356    }
357}
358
359#[cfg(test)]
360mod test {
361    use super::*;
362    use opendal_core::raw::oio::Read as _;
363    use opendal_core::raw::oio::ReadStream as _;
364    use opendal_core::raw::oio::Write as _;
365    use sqlx::SqlitePool;
366
367    async fn build_client() -> OnceCell<SqlitePool> {
368        let config = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
369        let pool = SqlitePool::connect_with(config).await.unwrap();
370        OnceCell::from_value(pool)
371    }
372
373    async fn build_backend() -> SqliteBackend {
374        let core = SqliteCore {
375            pool: build_client().await,
376            config: Default::default(),
377            table: "test_table".to_string(),
378            key_field: "key".to_string(),
379            value_field: "value".to_string(),
380        };
381
382        SqliteBackend::new(core)
383    }
384
385    #[tokio::test]
386    async fn test_sqlite_backend_creation() {
387        let backend = build_backend().await;
388
389        // Verify basic properties
390        assert_eq!(backend.root, "/");
391        assert_eq!(backend.info.scheme(), SQLITE_SCHEME);
392        assert!(backend.capability().read);
393        assert!(backend.capability().write);
394        assert!(backend.capability().delete);
395        assert!(backend.capability().stat);
396    }
397
398    #[tokio::test]
399    async fn test_sqlite_backend_with_root() {
400        let backend = build_backend()
401            .await
402            .with_normalized_root("/test/".to_string());
403
404        assert_eq!(backend.root, "/test/");
405        assert_eq!(backend.info.root(), Arc::from("/test/"));
406    }
407
408    #[tokio::test]
409    async fn test_sqlite_read_range_from_offset_reads_to_eof() {
410        let backend = build_backend().await;
411
412        let pool = backend.core.get_client().await.unwrap();
413        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value BLOB)")
414            .execute(pool)
415            .await
416            .unwrap();
417
418        let ctx = OperationContext::new();
419        let mut writer = backend.write(&ctx, "hello", OpWrite::default()).unwrap();
420        writer.write(Buffer::from("hello world")).await.unwrap();
421        writer.close().await.unwrap();
422
423        let reader = backend.read(&ctx, "hello", OpRead::default()).unwrap();
424        let (_, mut stream) = reader.open(BytesRange::from(6_u64..)).await.unwrap();
425        let buffer = stream.read_all().await.unwrap();
426
427        assert_eq!(buffer.to_vec(), b"world");
428    }
429
430    #[tokio::test]
431    async fn test_sqlite_stat_uses_value_length() {
432        let backend = build_backend().await;
433
434        let pool = backend.core.get_client().await.unwrap();
435        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value BLOB)")
436            .execute(pool)
437            .await
438            .unwrap();
439
440        let ctx = OperationContext::new();
441        let mut writer = backend.write(&ctx, "key_id", OpWrite::default()).unwrap();
442        writer.write(Buffer::from("hello world")).await.unwrap();
443        writer.close().await.unwrap();
444
445        let rp = backend
446            .stat(&ctx, "key_id", OpStat::default())
447            .await
448            .unwrap();
449
450        assert_eq!(rp.into_metadata().content_length(), 11);
451    }
452
453    #[tokio::test]
454    async fn test_sqlite_stat_returns_byte_length_for_text_value() {
455        let backend = build_backend().await;
456
457        let pool = backend.core.get_client().await.unwrap();
458        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value BLOB)")
459            .execute(pool)
460            .await
461            .unwrap();
462        sqlx::query("INSERT INTO test_table (key, value) VALUES ($1, $2)")
463            .bind("key_id")
464            .bind("你好")
465            .execute(pool)
466            .await
467            .unwrap();
468
469        let ctx = OperationContext::new();
470
471        let rp = backend
472            .stat(&ctx, "key_id", OpStat::default())
473            .await
474            .unwrap();
475        assert_eq!(rp.into_metadata().content_length(), 6);
476
477        let reader = backend.read(&ctx, "key_id", OpRead::default()).unwrap();
478        let (rp, mut stream) = reader.open(BytesRange::from(0_u64..3)).await.unwrap();
479        let buffer = stream.read_all().await.unwrap();
480
481        assert_eq!(rp.into_metadata().unwrap().content_length(), 6);
482        assert_eq!(buffer.to_vec(), "你".as_bytes());
483    }
484
485    #[tokio::test]
486    async fn test_sqlite_stat_returns_byte_length_for_text_column() {
487        let backend = build_backend().await;
488        let pool = backend.core.get_client().await.unwrap();
489
490        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value TEXT)")
491            .execute(pool)
492            .await
493            .unwrap();
494        sqlx::query("INSERT INTO test_table (key, value) VALUES ($1, $2)")
495            .bind("key_id")
496            .bind("你好")
497            .execute(pool)
498            .await
499            .unwrap();
500
501        let ctx = OperationContext::new();
502
503        let rp = backend
504            .stat(&ctx, "key_id", OpStat::default())
505            .await
506            .unwrap();
507        assert_eq!(rp.into_metadata().content_length(), 6);
508
509        let reader = backend.read(&ctx, "key_id", OpRead::default()).unwrap();
510        let (rp, mut stream) = reader.open(BytesRange::from(0_u64..3)).await.unwrap();
511        let buffer = stream.read_all().await.unwrap();
512
513        assert_eq!(rp.into_metadata().unwrap().content_length(), 6);
514        assert_eq!(buffer.to_vec(), "你".as_bytes());
515    }
516}
opendal_service_sqlite/backend.rs

opendal_service_sqlite/
backend.rs
