upgrade to df 53

Author: alexanderbianchi

quickwit/Cargo.lock

@@ -43,6 +43,9 @@ members = [
 
 # The following list excludes `quickwit-metastore-utils`
 # from the default member to ease build/deps.
+#
+# DataFusion crates are opt-in and covered by their own CI lane. Keep them out
+# of the default members so a plain `cargo build` does not pull DataFusion in.
 default-members = [
   "quickwit-actors",
   "quickwit-aws",
@@ -53,9 +56,7 @@ default-members = [
   "quickwit-common",
   "quickwit-config",
   "quickwit-control-plane",
-  "quickwit-datafusion",
   "quickwit-datetime",
-  "quickwit-df-core",
   "quickwit-directories",
   "quickwit-doc-mapper",
   "quickwit-index-management",

quickwit/Cargo.toml

@@ -8,10 +8,3 @@
 
 # openssl-macros doesn't declare a repository in its crate metadata
 "openssl-macros" = { origin = "https://github.com/sfackler/rust-openssl" }
-
-# datafusion-distributed 0.1.0 is pinned as a git dep; the Cargo.toml at that
-# rev does not declare a license field (added upstream after our pinned commit).
-# The crate is Apache-2.0 — see the crate's current Cargo.toml and
-# https://crates.io/crates/datafusion-distributed. Bumping to the crates.io
-# release requires a datafusion 52 → 53 workspace bump, tracked separately.
-"datafusion-distributed-0.1.0" = { license = "Apache-2.0", origin = "https://github.com/datafusion-contrib/datafusion-distributed" }

quickwit/license-tool.toml

@@ -35,19 +35,19 @@ quickwit-search = { workspace = true }
 quickwit-storage = { workspace = true }
 
 arrow = { workspace = true }
-datafusion = "52"
-datafusion-substrait = "52"
-datafusion-datasource = "52"
-datafusion-sql = "52"
-datafusion-physical-plan = "52"
-datafusion-datasource-parquet = "52"
-datafusion-distributed = { git = "https://github.com/datafusion-contrib/datafusion-distributed", rev = "0f2c8be3e148b0bd5c7f17b23f2df8bb1201d5fb" }
-object_store = "0.12"
+datafusion = "53"
+datafusion-substrait = "53"
+datafusion-datasource = "53"
+datafusion-sql = "53"
+datafusion-physical-plan = "53"
+datafusion-datasource-parquet = "53"
+datafusion-distributed = "1.0"
+object_store = "0.13"
 
 [dev-dependencies]
 bytesize = { workspace = true }
-datafusion = "52"
-datafusion-substrait = "52"
+datafusion = "53"
+datafusion-substrait = "53"
 prost = { workspace = true }
 serde_json = { workspace = true }
 tempfile = { workspace = true }

quickwit/quickwit-datafusion/Cargo.toml

@@ -26,12 +26,11 @@
 //! This registry resolves the tension by doing the sync/async split on
 //! the correct axis:
 //!
-//! - `get_store` (sync) builds a [`QuickwitObjectStore`] wrapper on demand.
-//!   Construction is cheap — just stashes the URI and a clone of the
-//!   resolver.
+//! - `get_store` (sync) builds a [`QuickwitObjectStore`] wrapper on demand. Construction is cheap —
+//!   just stashes the URI and a clone of the resolver.
 //! - `QuickwitObjectStore`'s own methods are already async, so the actual
-//!   `StorageResolver::resolve(&uri).await` runs the first time
-//!   DataFusion asks for data — inside the `ObjectStore` method itself.
+//!   `StorageResolver::resolve(&uri).await` runs the first time DataFusion asks for data — inside
+//!   the `ObjectStore` method itself.
 //!
 //! End result: one `StorageResolver` per node, used per-request, no
 //! global cache warm-up, no per-query listing. Matches the search-leaf
@@ -48,8 +47,7 @@ use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::{Arc, RwLock};
 
-use datafusion::common::DataFusionError;
-use datafusion::common::Result as DFResult;
+use datafusion::common::{DataFusionError, Result as DFResult};
 use datafusion::execution::object_store::{DefaultObjectStoreRegistry, ObjectStoreRegistry};
 use object_store::ObjectStore;
 use quickwit_common::uri::Uri;
@@ -140,10 +138,8 @@ impl ObjectStoreRegistry for QuickwitObjectStoreRegistry {
                 "failed to build Quickwit URI from `{key}`: {err}"
             ))))
         })?;
-        let store: Arc<dyn ObjectStore> = Arc::new(QuickwitObjectStore::new(
-            uri,
-            self.storage_resolver.clone(),
-        ));
+        let store: Arc<dyn ObjectStore> =
+            Arc::new(QuickwitObjectStore::new(uri, self.storage_resolver.clone()));
         let mut write = self
             .lazy_stores
             .write()

quickwit/quickwit-datafusion/src/object_store_registry.rs

@@ -28,9 +28,9 @@
 
 use std::sync::Arc;
 
-use datafusion::arrow as arrow;
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
+use datafusion::arrow;
 use datafusion::catalog::{Session, TableProviderFactory};
 use datafusion::error::{DataFusionError, Result as DFResult};
 use datafusion::logical_expr::CreateExternalTable;

quickwit/quickwit-datafusion/src/sources/metrics/factory.rs

@@ -183,12 +183,8 @@ impl SchemaProvider for MetricsSchemaProvider {
             return Ok(Some(provider));
         }
 
-        resolve_metrics_table_provider(
-            self.index_resolver.as_ref(),
-            name,
-            minimal_base_schema(),
-        )
-        .await
+        resolve_metrics_table_provider(self.index_resolver.as_ref(), name, minimal_base_schema())
+            .await
     }
 
     fn table_exist(&self, name: &str) -> bool {

quickwit/quickwit-datafusion/src/sources/metrics/mod.rs

@@ -21,10 +21,10 @@ use std::any::Any;
 use std::fmt;
 use std::sync::Arc;
 
-use datafusion::arrow as arrow;
 use arrow::compute::SortOptions;
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
+use datafusion::arrow;
 use datafusion::catalog::Session;
 use datafusion::datasource::TableProvider;
 use datafusion::datasource::source::DataSourceExec;
@@ -71,9 +71,8 @@ pub trait MetricsSplitProvider: Send + Sync + fmt::Debug {
 /// `ObjectStoreUrl` only accepts a scheme + authority, so the full index
 /// URI is split:
 /// - `scheme://authority` goes into the `ObjectStoreUrl`.
-/// - The path component (e.g. the `my-index/` part of
-///   `s3://bucket/my-index/`) is prepended to each split's filename in the
-///   emitted `PartitionedFile`s.
+/// - The path component (e.g. the `my-index/` part of `s3://bucket/my-index/`) is prepended to each
+///   split's filename in the emitted `PartitionedFile`s.
 ///
 /// Neither this type nor `scan()` constructs an `Arc<dyn ObjectStore>` —
 /// the session's custom

quickwit/quickwit-datafusion/src/sources/metrics/table_provider.rs

@@ -33,7 +33,7 @@ use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 use object_store::memory::InMemory;
 use object_store::path::Path as ObjectPath;
-use object_store::{ObjectStore, PutPayload};
+use object_store::{ObjectStoreExt, PutPayload};
 use quickwit_parquet_engine::split::{ParquetSplitId, ParquetSplitMetadata, TimeRange};
 use quickwit_parquet_engine::storage::{ParquetWriter, ParquetWriterConfig};
 use quickwit_parquet_engine::table_config::TableConfig;

quickwit/quickwit-datafusion/src/sources/metrics/test_utils.rs

@@ -26,19 +26,20 @@
 //! This mirrors how `quickwit-search` uses the resolver:
 //! `storage_resolver.resolve(&uri).await` per request, no global registry.
 //!
-//! Only read operations (`get_opts`, `get_range`, `head`) are implemented.
-//! All write and list operations return `NotSupported` — DataFusion only
-//! reads parquet files through this store.
+//! Only `get_opts` is implemented. All write, delete, copy, and list
+//! operations return `NotSupported` — DataFusion only reads parquet files
+//! through this store.
 
 use std::sync::Arc;
 
 use async_trait::async_trait;
 use bytes::Bytes;
+use futures::StreamExt;
 use futures::stream::BoxStream;
 use object_store::path::Path as ObjectPath;
 use object_store::{
-    GetOptions, GetRange, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta,
-    ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult,
+    CopyOptions, GetOptions, GetRange, GetResult, GetResultPayload, ListResult, MultipartUpload,
+    ObjectMeta, ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult,
     Result as ObjectStoreResult,
 };
 use quickwit_common::uri::Uri;
@@ -127,10 +128,13 @@ impl ObjectStore for QuickwitObjectStore {
         let location_str = location.as_ref();
         let map_err = |err| to_object_store_error(err, location_str);
 
-        let (bytes, byte_range) = match options.range {
+        let (bytes, byte_range) = match &options.range {
             Some(GetRange::Bounded(r)) => {
                 let usize_range = r.start as usize..r.end as usize;
-                let data = storage.get_slice(&path, usize_range).await.map_err(map_err)?;
+                let data = storage
+                    .get_slice(&path, usize_range)
+                    .await
+                    .map_err(map_err)?;
                 let b = Bytes::copy_from_slice(data.as_ref());
                 let len = b.len() as u64;
                 // The storage may return fewer bytes than requested if the range
@@ -140,17 +144,24 @@ impl ObjectStore for QuickwitObjectStore {
             }
             Some(GetRange::Suffix(n)) => {
                 let file_size = storage.file_num_bytes(&path).await.map_err(map_err)?;
-                let start = file_size.saturating_sub(n);
+                let start = file_size.saturating_sub(*n);
                 let usize_range = start as usize..file_size as usize;
-                let data = storage.get_slice(&path, usize_range).await.map_err(map_err)?;
+                let data = storage
+                    .get_slice(&path, usize_range)
+                    .await
+                    .map_err(map_err)?;
                 let b = Bytes::copy_from_slice(data.as_ref());
                 let len = b.len() as u64;
                 (b, start..start + len)
             }
             Some(GetRange::Offset(start)) => {
                 let file_size = storage.file_num_bytes(&path).await.map_err(map_err)?;
+                let start = *start;
                 let usize_range = start as usize..file_size as usize;
-                let data = storage.get_slice(&path, usize_range).await.map_err(map_err)?;
+                let data = storage
+                    .get_slice(&path, usize_range)
+                    .await
+                    .map_err(map_err)?;
                 let b = Bytes::copy_from_slice(data.as_ref());
                 let len = b.len() as u64;
                 (b, start..start + len)
@@ -171,45 +182,22 @@ impl ObjectStore for QuickwitObjectStore {
             e_tag: None,
             version: None,
         };
+        options.check_preconditions(&meta)?;
+
+        let payload = if options.head {
+            GetResultPayload::Stream(Box::pin(futures::stream::empty()))
+        } else {
+            GetResultPayload::Stream(Box::pin(futures::stream::once(async { Ok(bytes) })))
+        };
+
         Ok(GetResult {
-            payload: GetResultPayload::Stream(Box::pin(futures::stream::once(async { Ok(bytes) }))),
+            payload,
             meta,
             range: byte_range,
             attributes: Default::default(),
         })
     }
 
-    async fn get_range(
-        &self,
-        location: &ObjectPath,
-        range: std::ops::Range<u64>,
-    ) -> ObjectStoreResult<Bytes> {
-        let storage = self.storage().await?;
-        let path = object_path_to_std(location);
-        let usize_range = range.start as usize..range.end as usize;
-        let data = storage
-            .get_slice(&path, usize_range)
-            .await
-            .map_err(|err| to_object_store_error(err, location.as_ref()))?;
-        Ok(Bytes::copy_from_slice(data.as_ref()))
-    }
-
-    async fn head(&self, location: &ObjectPath) -> ObjectStoreResult<ObjectMeta> {
-        let storage = self.storage().await?;
-        let path = object_path_to_std(location);
-        let size = storage
-            .file_num_bytes(&path)
-            .await
-            .map_err(|err| to_object_store_error(err, location.as_ref()))?;
-        Ok(ObjectMeta {
-            location: location.clone(),
-            last_modified: chrono::Utc::now(),
-            size,
-            e_tag: None,
-            version: None,
-        })
-    }
-
     async fn put_opts(
         &self,
         _location: &ObjectPath,
@@ -231,10 +219,18 @@ impl ObjectStore for QuickwitObjectStore {
         })
     }
 
-    async fn delete(&self, _location: &ObjectPath) -> ObjectStoreResult<()> {
-        Err(object_store::Error::NotSupported {
-            source: "QuickwitObjectStore is read-only".into(),
-        })
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, ObjectStoreResult<ObjectPath>>,
+    ) -> BoxStream<'static, ObjectStoreResult<ObjectPath>> {
+        locations
+            .map(|location| match location {
+                Ok(_) => Err(object_store::Error::NotSupported {
+                    source: "QuickwitObjectStore is read-only".into(),
+                }),
+                Err(err) => Err(err),
+            })
+            .boxed()
     }
 
     fn list(
@@ -257,16 +253,11 @@ impl ObjectStore for QuickwitObjectStore {
         })
     }
 
-    async fn copy(&self, _from: &ObjectPath, _to: &ObjectPath) -> ObjectStoreResult<()> {
-        Err(object_store::Error::NotSupported {
-            source: "QuickwitObjectStore is read-only".into(),
-        })
-    }
-
-    async fn copy_if_not_exists(
+    async fn copy_opts(
         &self,
         _from: &ObjectPath,
         _to: &ObjectPath,
+        _options: CopyOptions,
     ) -> ObjectStoreResult<()> {
         Err(object_store::Error::NotSupported {
             source: "QuickwitObjectStore is read-only".into(),