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 mea::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
216    fn info(&self) -> ServiceInfo {
217        self.info.clone()
218    }
219
220    fn capability(&self) -> Capability {
221        self.capability
222    }
223
224    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
225        let p = build_abs_path(&self.root, path);
226
227        if p == build_abs_path(&self.root, "") {
228            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
229        } else {
230            let length = self.core.get_length(&p).await?;
231            match length {
232                Some(length) => Ok(RpStat::new(
233                    Metadata::new(EntryMode::from_path(&p)).with_content_length(length as u64),
234                )),
235                None => {
236                    // Check if this might be a directory by looking for keys with this prefix
237                    let dir_path = if p.ends_with('/') {
238                        p.clone()
239                    } else {
240                        format!("{}/", p)
241                    };
242                    let count = self.core.count_under(&dir_path).await?;
243
244                    if count > 0 {
245                        // Directory exists (has children)
246                        Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
247                    } else {
248                        Err(Error::new(ErrorKind::NotFound, "key not found in sqlite"))
249                    }
250                }
251            }
252        }
253    }
254    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
255        let output: oio::StreamReader<SqliteReader> = {
256            Ok(oio::StreamReader::new(SqliteReader::new(
257                self.clone(),
258                path,
259                args,
260            )))
261        }?;
262
263        Ok(output)
264    }
265
266    fn write(&self, _ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
267        let output: SqliteWriter = {
268            let p = build_abs_path(&self.root, path);
269            Ok(SqliteWriter::new(self.core.clone(), &p))
270        }?;
271
272        Ok(output)
273    }
274
275    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
276        let output: oio::OneShotDeleter<SqliteDeleter> = {
277            Ok(oio::OneShotDeleter::new(SqliteDeleter::new(
278                self.core.clone(),
279                self.root.clone(),
280            )))
281        }?;
282
283        Ok(output)
284    }
285
286    async fn create_dir(
287        &self,
288        _ctx: &OperationContext,
289        path: &str,
290        _: OpCreateDir,
291    ) -> Result<RpCreateDir> {
292        let p = build_abs_path(&self.root, path);
293
294        // Ensure path ends with '/' for directory marker
295        let dir_path = if p.ends_with('/') {
296            p
297        } else {
298            format!("{}/", p)
299        };
300
301        // Store directory marker with empty content
302        self.core.set(&dir_path, Buffer::new()).await?;
303
304        Ok(RpCreateDir::default())
305    }
306
307    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
308        Err(Error::new(
309            ErrorKind::Unsupported,
310            "operation is not supported",
311        ))
312    }
313
314    fn copy(
315        &self,
316        _ctx: &OperationContext,
317        _from: &str,
318        _to: &str,
319        _args: OpCopy,
320        _opts: OpCopier,
321    ) -> Result<Self::Copier> {
322        Err(Error::new(
323            ErrorKind::Unsupported,
324            "operation is not supported",
325        ))
326    }
327
328    async fn rename(
329        &self,
330        _ctx: &OperationContext,
331        _from: &str,
332        _to: &str,
333        _args: OpRename,
334    ) -> Result<RpRename> {
335        Err(Error::new(
336            ErrorKind::Unsupported,
337            "operation is not supported",
338        ))
339    }
340
341    async fn presign(
342        &self,
343        _ctx: &OperationContext,
344        _path: &str,
345        _args: OpPresign,
346    ) -> Result<RpPresign> {
347        Err(Error::new(
348            ErrorKind::Unsupported,
349            "operation is not supported",
350        ))
351    }
352}
353
354#[cfg(test)]
355mod test {
356    use super::*;
357    use opendal_core::raw::oio::Read as _;
358    use opendal_core::raw::oio::ReadStream as _;
359    use opendal_core::raw::oio::Write as _;
360    use sqlx::SqlitePool;
361
362    async fn build_client() -> OnceCell<SqlitePool> {
363        let config = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
364        let pool = SqlitePool::connect_with(config).await.unwrap();
365        OnceCell::from_value(pool)
366    }
367
368    async fn build_backend() -> SqliteBackend {
369        let core = SqliteCore {
370            pool: build_client().await,
371            config: Default::default(),
372            table: "test_table".to_string(),
373            key_field: "key".to_string(),
374            value_field: "value".to_string(),
375        };
376
377        SqliteBackend::new(core)
378    }
379
380    #[tokio::test]
381    async fn test_sqlite_backend_creation() {
382        let backend = build_backend().await;
383
384        // Verify basic properties
385        assert_eq!(backend.root, "/");
386        assert_eq!(backend.info.scheme(), SQLITE_SCHEME);
387        assert!(backend.capability().read);
388        assert!(backend.capability().write);
389        assert!(backend.capability().delete);
390        assert!(backend.capability().stat);
391    }
392
393    #[tokio::test]
394    async fn test_sqlite_backend_with_root() {
395        let backend = build_backend()
396            .await
397            .with_normalized_root("/test/".to_string());
398
399        assert_eq!(backend.root, "/test/");
400        assert_eq!(backend.info.root(), Arc::from("/test/"));
401    }
402
403    #[tokio::test]
404    async fn test_sqlite_read_range_from_offset_reads_to_eof() {
405        let backend = build_backend().await;
406
407        let pool = backend.core.get_client().await.unwrap();
408        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value BLOB)")
409            .execute(pool)
410            .await
411            .unwrap();
412
413        let ctx = OperationContext::new();
414        let mut writer = backend.write(&ctx, "hello", OpWrite::default()).unwrap();
415        writer.write(Buffer::from("hello world")).await.unwrap();
416        writer.close().await.unwrap();
417
418        let reader = backend.read(&ctx, "hello", OpRead::default()).unwrap();
419        let (_, mut stream) = reader.open(BytesRange::from(6_u64..)).await.unwrap();
420        let buffer = stream.read_all().await.unwrap();
421
422        assert_eq!(buffer.to_vec(), b"world");
423    }
424
425    #[tokio::test]
426    async fn test_sqlite_stat_uses_value_length() {
427        let backend = build_backend().await;
428
429        let pool = backend.core.get_client().await.unwrap();
430        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value BLOB)")
431            .execute(pool)
432            .await
433            .unwrap();
434
435        let ctx = OperationContext::new();
436        let mut writer = backend.write(&ctx, "key_id", OpWrite::default()).unwrap();
437        writer.write(Buffer::from("hello world")).await.unwrap();
438        writer.close().await.unwrap();
439
440        let rp = backend
441            .stat(&ctx, "key_id", OpStat::default())
442            .await
443            .unwrap();
444
445        assert_eq!(rp.into_metadata().content_length(), 11);
446    }
447
448    #[tokio::test]
449    async fn test_sqlite_stat_returns_byte_length_for_text_value() {
450        let backend = build_backend().await;
451
452        let pool = backend.core.get_client().await.unwrap();
453        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value BLOB)")
454            .execute(pool)
455            .await
456            .unwrap();
457        sqlx::query("INSERT INTO test_table (key, value) VALUES ($1, $2)")
458            .bind("key_id")
459            .bind("你好")
460            .execute(pool)
461            .await
462            .unwrap();
463
464        let ctx = OperationContext::new();
465
466        let rp = backend
467            .stat(&ctx, "key_id", OpStat::default())
468            .await
469            .unwrap();
470        assert_eq!(rp.into_metadata().content_length(), 6);
471
472        let reader = backend.read(&ctx, "key_id", OpRead::default()).unwrap();
473        let (rp, mut stream) = reader.open(BytesRange::from(0_u64..3)).await.unwrap();
474        let buffer = stream.read_all().await.unwrap();
475
476        assert_eq!(rp.into_metadata().unwrap().content_length(), 6);
477        assert_eq!(buffer.to_vec(), "你".as_bytes());
478    }
479
480    #[tokio::test]
481    async fn test_sqlite_stat_returns_byte_length_for_text_column() {
482        let backend = build_backend().await;
483        let pool = backend.core.get_client().await.unwrap();
484
485        sqlx::query("CREATE TABLE test_table (key TEXT PRIMARY KEY, value TEXT)")
486            .execute(pool)
487            .await
488            .unwrap();
489        sqlx::query("INSERT INTO test_table (key, value) VALUES ($1, $2)")
490            .bind("key_id")
491            .bind("你好")
492            .execute(pool)
493            .await
494            .unwrap();
495
496        let ctx = OperationContext::new();
497
498        let rp = backend
499            .stat(&ctx, "key_id", OpStat::default())
500            .await
501            .unwrap();
502        assert_eq!(rp.into_metadata().content_length(), 6);
503
504        let reader = backend.read(&ctx, "key_id", OpRead::default()).unwrap();
505        let (rp, mut stream) = reader.open(BytesRange::from(0_u64..3)).await.unwrap();
506        let buffer = stream.read_all().await.unwrap();
507
508        assert_eq!(rp.into_metadata().unwrap().content_length(), 6);
509        assert_eq!(buffer.to_vec(), "你".as_bytes());
510    }
511}
opendal_service_sqlite/backend.rs

opendal_service_sqlite/
backend.rs
