This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 380062ec build(deps): upgrade lance to 13 (#845)
380062ec is described below
commit 380062ecfb86abfa76d61570c31b61e70898ca9e
Author: Y Ethan Guo <[email protected]>
AuthorDate: Tue Oct 27 16:30:19 2026 -0700
build(deps): upgrade lance to 13 (#845)
Lance 13 is built on object_store 0.14 while the workspace is on 0.13, so
the Lance reader now hands Lance its store through an adapter implementing the
0.14 ObjectStore trait over the workspace's store.
---
crates/core/Cargo.toml | 19 +-
crates/core/src/file_group/base_file/lance.rs | 46 ++-
.../src/file_group/base_file/lance_object_store.rs | 435 +++++++++++++++++++++
crates/core/src/file_group/base_file/mod.rs | 1 +
4 files changed, 490 insertions(+), 11 deletions(-)
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index b1acb775..57efb12c 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -77,6 +77,7 @@ url = { workspace = true }
# runtime / async
async-recursion = { workspace = true }
+async-trait = { workspace = true }
dashmap = { workspace = true }
futures = { workspace = true }
tokio = { workspace = true }
@@ -92,17 +93,22 @@ flate2 = { workspace = true }
# prost-build at compile time, which shells out to `protoc`. Any build of
# hudi-core (CI, Docker, local dev, downstream consumers) must therefore
# have `protoc` available on PATH — there is no opt-out.
-lance-core = { version = "11.0.0" }
-lance-encoding = { version = "11.0.0" }
-lance-file = { version = "11.0.0" }
-lance-io = { version = "11.0.0" }
-# Lance 11 is built on arrow 58 while the workspace is on 59. These are the
+lance-core = { version = "13.0.0" }
+lance-encoding = { version = "13.0.0" }
+lance-file = { version = "13.0.0" }
+lance-io = { version = "13.0.0" }
+# Lance 13 is built on arrow 58 while the workspace is on 59. These are the
# arrow crates Lance's API speaks, renamed so the Lance reader can export its
# batches through the Arrow C data interface and import them as the
# workspace's arrow; see `file_group::base_file::lance_arrow`. Drop them once a
# Lance release shares the workspace's arrow major.
arrow-array-lance = { package = "arrow-array", version = "58", features =
["ffi"] }
arrow-schema-lance = { package = "arrow-schema", version = "58", features =
["ffi"] }
+# Lance 13 is built on object_store 0.14 while the workspace is on 0.13; the
+# Lance reader wraps the workspace's store in Lance's `ObjectStore` trait; see
+# `file_group::base_file::lance_object_store`. Drop it with the arrow crates
+# above once the majors line up.
+object_store-lance = { package = "object_store", version = "0.14.1",
default-features = false }
# datafusion
datafusion = { workspace = true, optional = true }
@@ -112,9 +118,6 @@ datafusion-physical-expr = { workspace = true, optional =
true }
percent-encoding = { workspace = true }
[dev-dependencies]
-# Needed only to implement `object_store::ObjectStore` for the test-only
counting
-# wrapper: that trait is declared with `#[async_trait]`, so an impl must be
too.
-async-trait = { workspace = true }
hudi-test = { path = "../test" }
serial_test = { workspace = true }
# Also a regular dependency, but only an optional one behind `spill-rocksdb`.
diff --git a/crates/core/src/file_group/base_file/lance.rs
b/crates/core/src/file_group/base_file/lance.rs
index 9f41a67e..787608a5 100644
--- a/crates/core/src/file_group/base_file/lance.rs
+++ b/crates/core/src/file_group/base_file/lance.rs
@@ -39,9 +39,11 @@ use lance_io::scheduler::{FileScheduler, ScanScheduler,
SchedulerConfig};
use lance_io::utils::CachedFileSize;
use object_store::ObjectStoreExt;
use object_store::path::Path as ObjPath;
+use object_store_lance::path::Path as LancePath;
use tokio::sync::OnceCell;
use super::lance_arrow;
+use super::lance_object_store::LanceObjectStoreAdapter;
use super::reader::{BaseFileReadOptions, BaseFileReader, BaseFileStream};
use crate::statistics::{ColumnStatistics, StatisticsContainer,
StatsGranularity};
use crate::storage::Storage;
@@ -84,14 +86,16 @@ impl LanceBaseFileReader {
),
);
// `ObjectStoreParams::object_store` is marked deprecated in
lance-io
- // 11.x in favor of implementing `ObjectStoreProvider`. We
still set
+ // in favor of implementing `ObjectStoreProvider`. We still set
// it here so Lance reuses the ObjectStore we already built
(with
// hudi-rs's storage options applied) and skips re-resolving
// credentials. `storage_options_accessor` itself is current
API.
#[allow(deprecated)]
let params = lance_io::object_store::ObjectStoreParams {
object_store: Some((
- self.storage.object_store.clone(),
+ Arc::new(LanceObjectStoreAdapter::new(
+ self.storage.object_store.clone(),
+ )),
(*self.storage.base_url).clone(),
)),
storage_options_accessor: Some(storage_accessor),
@@ -128,7 +132,8 @@ impl LanceBaseFileReader {
) -> Result<FileScheduler> {
let scheduler = self.scheduler().await?;
let obj_url = join_url_segments(&self.storage.base_url,
&[relative_path])?;
- let obj_path = ObjPath::from_url_path(obj_url.path())?;
+ let obj_path = LancePath::from_url_path(obj_url.path())
+ .map_err(|e| StorageError::InvalidPath(e.to_string()))?;
let cached_size = match known_size {
Some(size) => CachedFileSize::new(size),
None => CachedFileSize::unknown(),
@@ -394,6 +399,8 @@ impl BaseFileReader for LanceBaseFileReader {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::config::HudiConfigs;
+ use crate::storage::counting::CountingObjectStore;
use hudi_test::SampleTable;
use std::io::Write;
use tempfile::TempDir;
@@ -433,6 +440,39 @@ mod tests {
.to_string()
}
+ #[tokio::test]
+ async fn test_lance_reader_reads_data_through_a_non_local_store() {
+ // Lance reads a file:// URL straight from disk and uses the given
store
+ // only for size lookups, so only a non-local store carries the data
reads.
+ let file_name = lance_base_file_name();
+ let table_path = SampleTable::V9LanceNonpartitioned.path_to_cow();
+ let bytes =
std::fs::read(std::path::Path::new(&table_path).join(&file_name)).unwrap();
+ let memory = object_store::memory::InMemory::new();
+ memory
+ .put(&ObjPath::from(format!("table/{file_name}")), bytes.into())
+ .await
+ .unwrap();
+ let (store, counts) = CountingObjectStore::new(Arc::new(memory));
+ let storage = Storage::new_with_object_store(
+ Url::parse("memory:///table/").unwrap(),
+ store,
+ Arc::new(HudiConfigs::empty()),
+ );
+
+ let from_memory = LanceBaseFileReader::new(storage)
+ .read_data(&file_name, BaseFileReadOptions::new())
+ .await
+ .unwrap();
+ let from_disk = LanceBaseFileReader::new(lance_test_storage())
+ .read_data(&file_name, BaseFileReadOptions::new())
+ .await
+ .unwrap();
+
+ assert!(from_memory.num_rows() > 0);
+ assert_eq!(from_memory, from_disk);
+ assert!(counts.gets() > 0);
+ }
+
#[tokio::test]
async fn test_lance_reader_read_data_with_projection() {
let storage = lance_test_storage();
diff --git a/crates/core/src/file_group/base_file/lance_object_store.rs
b/crates/core/src/file_group/base_file/lance_object_store.rs
new file mode 100644
index 00000000..1016a0f6
--- /dev/null
+++ b/crates/core/src/file_group/base_file/lance_object_store.rs
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+//! Hands the workspace's object store to Lance, which is compiled against a
+//! different `object_store` major.
+//!
+//! Lance takes the store as a trait object of its own `object_store`, and to
+//! rustc the two crates' `ObjectStore` traits are unrelated, so the store is
+//! wrapped in an adapter that implements Lance's trait by converting each
+//! request and response. Both majors carry the same read types field for
+//! field. hudi-core only reads through Lance, so the write operations report
+//! `NotImplemented`.
+
+use std::fmt::{Display, Formatter};
+use std::ops::Range;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::stream::{self, BoxStream, StreamExt};
+use object_store::path::Path as HostPath;
+use object_store::{
+ Attribute as HostAttribute, Attributes as HostAttributes, Error as
HostError,
+ GetOptions as HostGetOptions, GetRange as HostGetRange, GetResult as
HostGetResult,
+ GetResultPayload as HostGetResultPayload, ObjectMeta as HostObjectMeta,
ObjectStore,
+};
+use object_store_lance::path::Path;
+use object_store_lance::{
+ Attribute, AttributeValue, Attributes, CopyOptions, Error, GetOptions,
GetRange, GetResult,
+ GetResultPayload, ListResult, MultipartUpload, ObjectMeta,
PutMultipartOptions, PutOptions,
+ PutPayload, PutResult, Result,
+};
+
+const STORE: &str = "LanceObjectStoreAdapter";
+
+/// Implements Lance's `ObjectStore` over the workspace's store.
+#[derive(Debug)]
+pub(super) struct LanceObjectStoreAdapter {
+ inner: Arc<dyn ObjectStore>,
+}
+
+impl LanceObjectStoreAdapter {
+ pub(super) fn new(inner: Arc<dyn ObjectStore>) -> Self {
+ Self { inner }
+ }
+}
+
+impl Display for LanceObjectStoreAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{}", self.inner)
+ }
+}
+
+#[async_trait]
+impl object_store_lance::ObjectStore for LanceObjectStoreAdapter {
+ async fn put_opts(
+ &self,
+ _location: &Path,
+ _payload: PutPayload,
+ _opts: PutOptions,
+ ) -> Result<PutResult> {
+ Err(not_implemented("put_opts"))
+ }
+
+ async fn put_multipart_opts(
+ &self,
+ _location: &Path,
+ _opts: PutMultipartOptions,
+ ) -> Result<Box<dyn MultipartUpload>> {
+ Err(not_implemented("put_multipart_opts"))
+ }
+
+ async fn get_opts(&self, location: &Path, options: GetOptions) ->
Result<GetResult> {
+ let result = self
+ .inner
+ .get_opts(&to_host_path(location)?, to_host_get_options(options))
+ .await
+ .map_err(to_lance_error)?;
+ to_lance_get_result(result)
+ }
+
+ async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) ->
Result<Vec<Bytes>> {
+ self.inner
+ .get_ranges(&to_host_path(location)?, ranges)
+ .await
+ .map_err(to_lance_error)
+ }
+
+ fn delete_stream(
+ &self,
+ locations: BoxStream<'static, Result<Path>>,
+ ) -> BoxStream<'static, Result<Path>> {
+ locations
+ .map(|_| Err(not_implemented("delete_stream")))
+ .boxed()
+ }
+
+ fn list(&self, prefix: Option<&Path>) -> BoxStream<'static,
Result<ObjectMeta>> {
+ match prefix.map(to_host_path).transpose() {
+ Ok(prefix) =>
to_lance_meta_stream(self.inner.list(prefix.as_ref())),
+ Err(e) => stream::once(async { Err(e) }).boxed(),
+ }
+ }
+
+ fn list_with_offset(
+ &self,
+ prefix: Option<&Path>,
+ offset: &Path,
+ ) -> BoxStream<'static, Result<ObjectMeta>> {
+ let paths = prefix
+ .map(to_host_path)
+ .transpose()
+ .and_then(|prefix| Ok((prefix, to_host_path(offset)?)));
+ match paths {
+ Ok((prefix, offset)) => {
+
to_lance_meta_stream(self.inner.list_with_offset(prefix.as_ref(), &offset))
+ }
+ Err(e) => stream::once(async { Err(e) }).boxed(),
+ }
+ }
+
+ async fn list_with_delimiter(&self, prefix: Option<&Path>) ->
Result<ListResult> {
+ let prefix = prefix.map(to_host_path).transpose()?;
+ let result = self
+ .inner
+ .list_with_delimiter(prefix.as_ref())
+ .await
+ .map_err(to_lance_error)?;
+ Ok(ListResult {
+ common_prefixes: result
+ .common_prefixes
+ .iter()
+ .map(to_lance_path)
+ .collect::<Result<_>>()?,
+ objects: result
+ .objects
+ .into_iter()
+ .map(to_lance_meta)
+ .collect::<Result<_>>()?,
+ extensions: Default::default(),
+ })
+ }
+
+ async fn copy_opts(&self, _from: &Path, _to: &Path, _options: CopyOptions)
-> Result<()> {
+ Err(not_implemented("copy_opts"))
+ }
+}
+
+fn not_implemented(operation: &str) -> Error {
+ Error::NotImplemented {
+ operation: operation.to_string(),
+ implementer: STORE.to_string(),
+ }
+}
+
+fn to_host_path(path: &Path) -> Result<HostPath> {
+ HostPath::parse(path.as_ref()).map_err(|e| Error::Generic {
+ store: STORE,
+ source: Box::new(e),
+ })
+}
+
+fn to_lance_path(path: &HostPath) -> Result<Path> {
+ Ok(Path::parse(path.as_ref())?)
+}
+
+fn to_host_get_options(options: GetOptions) -> HostGetOptions {
+ HostGetOptions {
+ if_match: options.if_match,
+ if_none_match: options.if_none_match,
+ if_modified_since: options.if_modified_since,
+ if_unmodified_since: options.if_unmodified_since,
+ range: options.range.map(|range| match range {
+ GetRange::Bounded(r) => HostGetRange::Bounded(r),
+ GetRange::Offset(o) => HostGetRange::Offset(o),
+ GetRange::Suffix(s) => HostGetRange::Suffix(s),
+ }),
+ version: options.version,
+ head: options.head,
+ extensions: options.extensions,
+ }
+}
+
+fn to_lance_get_result(result: HostGetResult) -> Result<GetResult> {
+ let payload = match result.payload {
+ HostGetResultPayload::File(file, path) => GetResultPayload::File(file,
path),
+ HostGetResultPayload::Stream(s) => {
+ GetResultPayload::Stream(s.map(|r|
r.map_err(to_lance_error)).boxed())
+ }
+ };
+ Ok(GetResult {
+ payload,
+ meta: to_lance_meta(result.meta)?,
+ range: result.range,
+ attributes: to_lance_attributes(&result.attributes),
+ extensions: Default::default(),
+ })
+}
+
+fn to_lance_meta(meta: HostObjectMeta) -> Result<ObjectMeta> {
+ Ok(ObjectMeta {
+ location: to_lance_path(&meta.location)?,
+ last_modified: meta.last_modified,
+ size: meta.size,
+ e_tag: meta.e_tag,
+ version: meta.version,
+ })
+}
+
+fn to_lance_meta_stream(
+ metas: BoxStream<'static, object_store::Result<HostObjectMeta>>,
+) -> BoxStream<'static, Result<ObjectMeta>> {
+ metas
+ .map(|r| r.map_err(to_lance_error).and_then(to_lance_meta))
+ .boxed()
+}
+
+fn to_lance_attributes(attributes: &HostAttributes) -> Attributes {
+ attributes
+ .iter()
+ .filter_map(|(key, value)| {
+ let key = match key {
+ HostAttribute::ContentDisposition =>
Attribute::ContentDisposition,
+ HostAttribute::ContentEncoding => Attribute::ContentEncoding,
+ HostAttribute::ContentLanguage => Attribute::ContentLanguage,
+ HostAttribute::ContentType => Attribute::ContentType,
+ HostAttribute::CacheControl => Attribute::CacheControl,
+ HostAttribute::StorageClass => Attribute::StorageClass,
+ HostAttribute::Metadata(name) =>
Attribute::Metadata(name.clone()),
+ _ => return None,
+ };
+ Some((key, AttributeValue::from(value.to_string())))
+ })
+ .collect()
+}
+
+/// Keeps the variants Lance tells apart (a missing object above all) and
+/// carries every other error as `Generic` with the original as its source.
+fn to_lance_error(error: HostError) -> Error {
+ match error {
+ HostError::NotFound { path, source } => Error::NotFound { path, source
},
+ HostError::AlreadyExists { path, source } => Error::AlreadyExists {
path, source },
+ HostError::Precondition { path, source } => Error::Precondition {
path, source },
+ HostError::NotModified { path, source } => Error::NotModified { path,
source },
+ HostError::NotSupported { source } => Error::NotSupported { source },
+ HostError::PermissionDenied { path, source } =>
Error::PermissionDenied { path, source },
+ HostError::Unauthenticated { path, source } => Error::Unauthenticated
{ path, source },
+ HostError::NotImplemented {
+ operation,
+ implementer,
+ } => Error::NotImplemented {
+ operation,
+ implementer,
+ },
+ HostError::Generic { store, source } => Error::Generic { store, source
},
+ other => Error::Generic {
+ store: STORE,
+ source: Box::new(other),
+ },
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use object_store::memory::InMemory;
+ use object_store::{ObjectStoreExt as HostObjectStoreExt, PutPayload as
HostPutPayload};
+ use object_store_lance::{ObjectStore as LanceObjectStore, ObjectStoreExt};
+
+ async fn adapter_over(files: &[(&str, &'static [u8])]) ->
LanceObjectStoreAdapter {
+ let inner = InMemory::new();
+ for (path, data) in files {
+ inner
+ .put(&HostPath::from(*path), HostPutPayload::from_static(data))
+ .await
+ .unwrap();
+ }
+ LanceObjectStoreAdapter::new(Arc::new(inner))
+ }
+
+ #[tokio::test]
+ async fn reads_whole_objects_ranges_and_metadata() {
+ let store = adapter_over(&[("dir/a.lance", b"0123456789")]).await;
+ let path = Path::from("dir/a.lance");
+
+ let whole = store.get(&path).await.unwrap();
+ assert_eq!(whole.meta.size, 10);
+ assert_eq!(whole.range, 0..10);
+ assert_eq!(
+ whole.bytes().await.unwrap(),
+ Bytes::from_static(b"0123456789")
+ );
+
+ assert_eq!(
+ store.get_range(&path, 2..5).await.unwrap(),
+ Bytes::from_static(b"234")
+ );
+ assert_eq!(
+ store.get_ranges(&path, &[0..1, 8..10]).await.unwrap(),
+ vec![Bytes::from_static(b"0"), Bytes::from_static(b"89")]
+ );
+
+ let suffix = store
+ .get_opts(
+ &path,
+ GetOptions {
+ range: Some(GetRange::Suffix(3)),
+ ..Default::default()
+ },
+ )
+ .await
+ .unwrap();
+ assert_eq!(suffix.range, 7..10);
+
+ let offset = store
+ .get_opts(
+ &path,
+ GetOptions {
+ range: Some(GetRange::Offset(6)),
+ ..Default::default()
+ },
+ )
+ .await
+ .unwrap();
+ assert_eq!(offset.range, 6..10);
+ assert_eq!(offset.bytes().await.unwrap(), Bytes::from_static(b"6789"));
+
+ let head = store.head(&path).await.unwrap();
+ assert_eq!(head.location, path);
+ assert_eq!(head.size, 10);
+ }
+
+ #[tokio::test]
+ async fn reads_local_files_through_the_file_payload() {
+ let dir = tempfile::tempdir().unwrap();
+ std::fs::write(dir.path().join("a.lance"), b"0123456789").unwrap();
+ let inner =
object_store::local::LocalFileSystem::new_with_prefix(dir.path()).unwrap();
+ let store = LanceObjectStoreAdapter::new(Arc::new(inner));
+ let path = Path::from("a.lance");
+
+ let result = store.get(&path).await.unwrap();
+ assert!(matches!(result.payload, GetResultPayload::File(..)));
+ assert_eq!(
+ result.bytes().await.unwrap(),
+ Bytes::from_static(b"0123456789")
+ );
+ assert_eq!(
+ store.get_range(&path, 4..6).await.unwrap(),
+ Bytes::from_static(b"45")
+ );
+ }
+
+ #[tokio::test]
+ async fn lists_with_and_without_delimiter() {
+ let store = adapter_over(&[
+ ("dir/a.lance", b"a"),
+ ("dir/b.lance", b"b"),
+ ("dir/sub/c.lance", b"c"),
+ ])
+ .await;
+ let prefix = Path::from("dir");
+
+ let mut all: Vec<_> = store
+ .list(Some(&prefix))
+ .map(|r| r.unwrap().location.to_string())
+ .collect()
+ .await;
+ all.sort();
+ assert_eq!(all, ["dir/a.lance", "dir/b.lance", "dir/sub/c.lance"]);
+
+ let after: Vec<_> = store
+ .list_with_offset(Some(&prefix), &Path::from("dir/a.lance"))
+ .map(|r| r.unwrap().location.to_string())
+ .collect()
+ .await;
+ assert!(!after.contains(&"dir/a.lance".to_string()));
+ assert_eq!(after.len(), 2);
+
+ let listed = store.list_with_delimiter(Some(&prefix)).await.unwrap();
+ assert_eq!(listed.common_prefixes, vec![Path::from("dir/sub")]);
+ let mut objects: Vec<_> = listed
+ .objects
+ .iter()
+ .map(|m| m.location.to_string())
+ .collect();
+ objects.sort();
+ assert_eq!(objects, ["dir/a.lance", "dir/b.lance"]);
+ }
+
+ #[tokio::test]
+ async fn keeps_not_found_distinct() {
+ let store = adapter_over(&[]).await;
+ let err = store.head(&Path::from("missing.lance")).await.unwrap_err();
+ assert!(matches!(err, Error::NotFound { .. }), "{err:?}");
+ }
+
+ #[tokio::test]
+ async fn rejects_writes() {
+ let store = adapter_over(&[]).await;
+ let path = Path::from("a.lance");
+
+ let errors = [
+ store
+ .put(&path, PutPayload::from_static(b"x"))
+ .await
+ .err()
+ .unwrap(),
+ store.put_multipart(&path).await.err().unwrap(),
+ store
+ .copy(&path, &Path::from("b.lance"))
+ .await
+ .err()
+ .unwrap(),
+ store.delete(&path).await.err().unwrap(),
+ ];
+ for err in errors {
+ assert!(matches!(err, Error::NotImplemented { .. }), "{err:?}");
+ }
+ }
+}
diff --git a/crates/core/src/file_group/base_file/mod.rs
b/crates/core/src/file_group/base_file/mod.rs
index 32a8e3fd..e72619d3 100644
--- a/crates/core/src/file_group/base_file/mod.rs
+++ b/crates/core/src/file_group/base_file/mod.rs
@@ -19,6 +19,7 @@
pub mod hfile;
pub mod lance;
mod lance_arrow;
+mod lance_object_store;
pub mod parquet;
pub mod reader;
