feat: Apache OpenDAL™ compatible backends

Cargo.toml

@@ -304,6 +304,7 @@ num-derive = "0.4.2"
 num-format = "0.4.4"
 num-traits = "0.2"
 numpy = "0.27"
+opendal = {version = "0.51", default-features = false}
 opentelemetry = {version = "0.31", features = ["trace", "metrics", "logs"]}
 opentelemetry-otlp = {version = "0.31", features = ["grpc-tonic", "logs"]}
 opentelemetry_sdk = {version = "0.31", features = ["logs"]}

daft/daft/__init__.pyi

@@ -903,6 +903,7 @@ class IOConfig:
     disable_suffix_range: bool
     tos: TosConfig
     gravitino: GravitinoConfig
+    backends: dict[str, dict[str, str]]
 
     def __init__(
         self,
@@ -915,6 +916,7 @@ class IOConfig:
         disable_suffix_range: bool | None = None,
         tos: TosConfig | None = None,
         gravitino: GravitinoConfig | None = None,
+        backends: dict[str, dict[str, str]] | None = None,
     ): ...
     def replace(
         self,
@@ -927,6 +929,7 @@ class IOConfig:
         disable_suffix_range: bool | None = None,
         tos: TosConfig | None = None,
         gravitino: GravitinoConfig | None = None,
+        backends: dict[str, dict[str, str]] | None = None,
     ) -> IOConfig:
         """Replaces values if provided, returning a new IOConfig."""
         ...

src/common/io-config/src/config.rs

@@ -1,4 +1,7 @@
-use std::fmt::{Display, Formatter};
+use std::{
+    collections::BTreeMap,
+    fmt::{Display, Formatter},
+};
 
 use serde::{Deserialize, Serialize};
 
@@ -18,6 +21,9 @@ pub struct IOConfig {
     /// disable suffix range requests, please use range with offset
     pub disable_suffix_range: bool,
     pub tos: TosConfig,
+    /// Additional backends configured via OpenDAL.
+    /// Keys are scheme names (e.g. "oss", "cos"), values are key-value config maps.
+    pub backends: BTreeMap<String, BTreeMap<String, String>>,
 }
 
 impl IOConfig {
@@ -60,6 +66,9 @@ impl IOConfig {
             "TOS config = {{ {} }}",
             self.tos.multiline_display().join(", ")
         ));
+        if !self.backends.is_empty() {
+            res.push(format!("OpenDAL backends = {:?}", self.backends));
+        }
         res
     }
 }

src/common/io-config/src/python.rs

@@ -1,5 +1,6 @@
 use std::{
     any::Any,
+    collections::HashMap,
     hash::{Hash, Hasher},
     sync::Arc,
 };
@@ -198,7 +199,8 @@ impl IOConfig {
         hf=None,
         disable_suffix_range=None,
         tos=None,
-        gravitino=None
+        gravitino=None,
+        backends=None
     ))]
     #[allow(clippy::too_many_arguments)]
     pub fn new(
@@ -211,6 +213,7 @@ impl IOConfig {
         disable_suffix_range: Option<bool>,
         tos: Option<TosConfig>,
         gravitino: Option<GravitinoConfig>,
+        backends: Option<HashMap<String, HashMap<String, String>>>,
     ) -> Self {
         Self {
             config: config::IOConfig {
@@ -223,6 +226,11 @@ impl IOConfig {
                 disable_suffix_range: disable_suffix_range.unwrap_or_default(),
                 tos: tos.unwrap_or_default().config,
                 gravitino: gravitino.unwrap_or_default().config,
+                backends: backends
+                    .unwrap_or_default()
+                    .into_iter()
+                    .map(|(k, v)| (k, v.into_iter().collect()))
+                    .collect(),
             },
         }
     }
@@ -238,7 +246,8 @@ impl IOConfig {
         hf=None,
         disable_suffix_range=None,
         tos=None,
-        gravitino=None
+        gravitino=None,
+        backends=None
     ))]
     #[allow(clippy::too_many_arguments)]
     pub fn replace(
@@ -252,6 +261,7 @@ impl IOConfig {
         disable_suffix_range: Option<bool>,
         tos: Option<TosConfig>,
         gravitino: Option<GravitinoConfig>,
+        backends: Option<HashMap<String, HashMap<String, String>>>,
     ) -> Self {
         Self {
             config: config::IOConfig {
@@ -281,6 +291,13 @@ impl IOConfig {
                 gravitino: gravitino
                     .map(|gravitino| gravitino.config)
                     .unwrap_or_else(|| self.config.gravitino.clone()),
+                backends: backends
+                    .map(|b| {
+                        b.into_iter()
+                            .map(|(k, v)| (k, v.into_iter().collect()))
+                            .collect()
+                    })
+                    .unwrap_or_else(|| self.config.backends.clone()),
             },
         }
     }
@@ -349,6 +366,22 @@ impl IOConfig {
         })
     }
 
+    /// Additional backends configured via OpenDAL
+    #[getter]
+    pub fn backends(&self) -> PyResult<HashMap<String, HashMap<String, String>>> {
+        Ok(self
+            .config
+            .backends
+            .iter()
+            .map(|(k, v)| {
+                (
+                    k.clone(),
+                    v.iter().map(|(k2, v2)| (k2.clone(), v2.clone())).collect(),
+                )
+            })
+            .collect())
+    }
+
     pub fn __hash__(&self) -> PyResult<u64> {
         use std::{collections::hash_map::DefaultHasher, hash::Hash};
 

src/daft-io/Cargo.toml

@@ -24,6 +24,14 @@ google-cloud-token = {version = "0.1.2"}
 home = "0.5.12"
 itertools = {workspace = true}
 log = {workspace = true}
+opendal = {workspace = true, features = [
+  "services-oss",
+  "services-cos",
+  "services-obs",
+  "services-memory",
+  "services-fs",
+  "services-github"
+]}
 pyo3 = {workspace = true, optional = true}
 rand = "0.8.5"
 regex = {version = "1.12.2"}