From 4c12ea1aba301b96a317e2b38674b5a5158dbe73 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Mon, 7 Sep 2026 08:16:47 +0000 Subject: [PATCH] chore(deps): bump opendal to 0.58.1 (#8742) * chore(deps): bump opendal to 0.58.1 Upgrade direct opendal dependency and workspace object_store_opendal pin from 0.57 to 0.58 (lockfile resolves opendal 0.58.1 / object_store_opendal 0.58.0). Adapt to OpenDAL 0.58 composition API: - Operator::new returns a finished operator; drop .finish() call sites - Replace HttpClientLayer / raw::HttpClient with OperationContext + HttpTransporter (ReqwestTransport) - Migrate SecureFsBackend and MockLayer from Access/LayeredAccess to Service + Layer::apply_service - Rewrite SecureFs reader/writer/lister for sync factories and StreamRead - Use OperatorInfo::capability() instead of removed native_capability() Signed-off-by: Xuanwo Signed-off-by: WenyXu * chore: retrigger CI after udeps runner segfault Signed-off-by: Xuanwo Signed-off-by: WenyXu * fix(object-store): restore suffix read simulation for secure fs Signed-off-by: WenyXu * fix: adapt remaining callers to OpenDAL 0.58 Signed-off-by: WenyXu --------- Signed-off-by: Xuanwo Signed-off-by: WenyXu Co-authored-by: WenyXu --- Cargo.lock | 308 +++++++------ Cargo.toml | 2 +- src/cli/src/common/object_store.rs | 6 +- src/cli/src/data/snapshot_storage.rs | 24 +- .../src/bin/query_perf_fixture/direct_sst.rs | 3 +- .../bin/query_perf_fixture/inspect_footer.rs | 4 +- .../query_regression_runner/materialize.rs | 4 +- src/cmd/src/datanode/parquetbench.rs | 3 +- .../datasource/src/object_store/azblob.rs | 4 +- src/common/datasource/src/object_store/gcs.rs | 4 +- src/common/datasource/src/object_store/oss.rs | 4 +- src/common/datasource/src/object_store/s3.rs | 4 +- src/common/datasource/src/test_util.rs | 4 +- src/common/meta/src/snapshot.rs | 2 +- src/common/procedure/src/local.rs | 2 +- src/common/procedure/src/store.rs | 2 +- src/common/procedure/src/store/state_store.rs | 4 +- src/file-engine/src/engine.rs | 5 +- src/file-engine/src/test_util.rs | 2 +- src/log-store/src/kafka/index/collector.rs | 5 +- src/metric-engine/src/test_util.rs | 2 +- src/mito2/src/access_layer.rs | 2 +- src/mito2/src/cache/file_cache.rs | 2 +- src/mito2/src/cache/manifest_cache.rs | 2 +- src/mito2/src/cache/test_util.rs | 2 +- src/mito2/src/cache/write_cache.rs | 6 +- src/mito2/src/gc/worker_test.rs | 22 +- src/mito2/src/manifest/storage.rs | 2 +- src/mito2/src/manifest/storage/staging.rs | 2 +- src/mito2/src/manifest/tests/checkpoint.rs | 10 +- src/mito2/src/region.rs | 2 +- src/mito2/src/region/opener.rs | 18 +- src/mito2/src/series_index/searcher.rs | 2 +- src/mito2/src/series_index/writer.rs | 2 +- src/mito2/src/sst/file_purger.rs | 4 +- src/mito2/src/sst/index.rs | 2 +- .../src/sst/index/bloom_filter/applier.rs | 6 +- .../sst/index/bloom_filter/applier/builder.rs | 2 +- .../src/sst/index/bloom_filter/creator.rs | 2 +- .../src/sst/index/fulltext_index/creator.rs | 2 +- .../src/sst/index/inverted_index/applier.rs | 6 +- .../index/inverted_index/applier/builder.rs | 2 +- .../src/sst/index/inverted_index/creator.rs | 2 +- src/mito2/src/sst/index/puffin_manager.rs | 2 +- src/mito2/src/sst/index/store.rs | 2 +- .../src/sst/index/vector_index/applier.rs | 2 +- src/mito2/src/sst/parquet/metadata.rs | 2 +- src/mito2/src/sst/parquet/reader.rs | 8 +- src/mito2/src/sst/range_index/searcher.rs | 2 +- src/mito2/src/sst/range_index/writer.rs | 2 +- src/mito2/src/test_util.rs | 16 +- src/mito2/src/test_util/scheduler_util.rs | 2 +- src/object-store/Cargo.toml | 3 +- src/object-store/src/factory.rs | 31 +- src/object-store/src/layers/mock.rs | 257 +++++------ src/object-store/src/lib.rs | 5 +- src/object-store/src/manager.rs | 2 +- src/object-store/src/secure_fs.rs | 414 +++++++++++------- src/object-store/src/util.rs | 17 +- src/object-store/tests/object_store_test.rs | 14 +- src/operator/src/statement/copy_database.rs | 2 +- tests-integration/src/test_util.rs | 8 +- 62 files changed, 705 insertions(+), 586 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1b0b338d33..be56131fce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1044,7 +1044,7 @@ dependencies = [ "bytes", "form_urlencoded", "futures-util", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "hyper 1.6.0", @@ -1094,7 +1094,7 @@ checksum = "68464cd0412f486726fb3373129ef5d2993f90c34bc2bc1c1e9943b2f4fc7ca6" dependencies = [ "bytes", "futures-core", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "mime", @@ -1117,7 +1117,7 @@ dependencies = [ "bytes", "futures-util", "headers", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "mime", @@ -1623,7 +1623,7 @@ dependencies = [ "maybe-owned", "rustix 1.0.7", "rustix-linux-procfs", - "windows-sys 0.61.2", + "windows-sys 0.60.2", "winx", ] @@ -2401,7 +2401,7 @@ name = "common-error" version = "1.3.0-alpha.1" dependencies = [ "common-macro", - "http 1.3.1", + "http 1.5.0", "serde", "snafu 0.8.6", "strum 0.27.1", @@ -3284,6 +3284,16 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" +[[package]] +name = "crc-fast" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5" +dependencies = [ + "digest 0.10.7", + "spin 0.10.1", +] + [[package]] name = "crc32c" version = "0.6.8" @@ -5181,7 +5191,7 @@ version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8acfe553027cd07fc5fafa81a84f19a7a87eaffaccd2162b6db05e8d6ce98084" dependencies = [ - "http 1.3.1", + "http 1.5.0", "prost 0.14.1", "tokio", "tokio-stream", @@ -5485,7 +5495,7 @@ dependencies = [ "get-size2", "greptime-proto", "hostname 0.4.1", - "http 1.3.1", + "http 1.5.0", "humantime-serde", "itertools 0.14.0", "lazy_static", @@ -5544,7 +5554,7 @@ checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" dependencies = [ "futures-core", "futures-sink", - "spin", + "spin 0.9.8", ] [[package]] @@ -6138,7 +6148,7 @@ dependencies = [ "fnv", "futures-core", "futures-sink", - "http 1.3.1", + "http 1.5.0", "indexmap 2.13.0", "slab", "tokio", @@ -6279,7 +6289,7 @@ dependencies = [ "base64 0.22.1", "bytes", "headers-core", - "http 1.3.1", + "http 1.5.0", "httpdate", "mime", "sha1 0.10.6", @@ -6291,7 +6301,7 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4" dependencies = [ - "http 1.3.1", + "http 1.5.0", ] [[package]] @@ -6405,12 +6415,11 @@ dependencies = [ [[package]] name = "http" -version = "1.3.1" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4a85d31aea989eead29a3aaf9e1115a180df8282431156e533de47660892565" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" dependencies = [ "bytes", - "fnv", "itoa", ] @@ -6432,7 +6441,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" dependencies = [ "bytes", - "http 1.3.1", + "http 1.5.0", ] [[package]] @@ -6443,7 +6452,7 @@ checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" dependencies = [ "bytes", "futures-core", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "pin-project-lite", ] @@ -6541,7 +6550,7 @@ dependencies = [ "futures-channel", "futures-util", "h2 0.4.11", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "httparse", "httpdate", @@ -6561,7 +6570,7 @@ dependencies = [ "bytes", "futures-util", "headers", - "http 1.3.1", + "http 1.5.0", "hyper 1.6.0", "hyper-rustls", "hyper-util", @@ -6578,7 +6587,7 @@ version = "0.27.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58" dependencies = [ - "http 1.3.1", + "http 1.5.0", "hyper 1.6.0", "hyper-util", "log", @@ -6627,7 +6636,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "hyper 1.6.0", "ipnet", @@ -7559,7 +7568,7 @@ dependencies = [ "either", "futures", "home", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "hyper 1.6.0", @@ -7593,7 +7602,7 @@ checksum = "40fb9bd8141cbc0fe6b0d9112d371679b4cb607b45c31dd68d92e40864a12975" dependencies = [ "chrono", "form_urlencoded", - "http 1.3.1", + "http 1.5.0", "json-patch", "k8s-openapi", "schemars 0.8.22", @@ -7730,7 +7739,7 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" dependencies = [ - "spin", + "spin 0.9.8", ] [[package]] @@ -8012,9 +8021,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.29" +version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" [[package]] name = "log-query" @@ -8756,11 +8765,11 @@ dependencies = [ "bytes", "encoding_rs", "futures-util", - "http 1.3.1", + "http 1.5.0", "httparse", "memchr", "mime", - "spin", + "spin 0.9.8", "version_check", ] @@ -9164,7 +9173,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -9472,6 +9481,7 @@ dependencies = [ "object_store", "object_store_opendal", "opendal", + "opendal-http-transport-reqwest", "prometheus 0.14.0", "rand 0.9.4", "reqwest 0.13.4", @@ -9495,7 +9505,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http 1.3.1", + "http 1.5.0", "humantime", "itertools 0.14.0", "parking_lot 0.12.4", @@ -9511,9 +9521,9 @@ dependencies = [ [[package]] name = "object_store_opendal" -version = "0.57.0" +version = "0.58.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eb12a624a41fce745838d0ef3701ff6c47797c13cd18ad3612fd2a3134fdbd8" +checksum = "88f165780495c17aa3ce86846600504198c3fffd99073521552751c2430fa6ac" dependencies = [ "async-trait", "bytes", @@ -9600,12 +9610,13 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "opendal" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96c9c85ce253ff87225e7669979d877a20c98a06604ec9d6dd5f4473e08f1ae1" +checksum = "4f20562cc7447fcc915fc5c23df305a412ea80a733c9f2fd9e2d267e2815be6d" dependencies = [ "ctor", "opendal-core", + "opendal-http-transport-reqwest", "opendal-layer-concurrent-limit", "opendal-layer-logging", "opendal-layer-prometheus", @@ -9623,24 +9634,22 @@ dependencies = [ [[package]] name = "opendal-core" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4f8607c90e2c963a91467f50fb49fbc7fb3d573f88cea219ca59ccd3740b309" +checksum = "ec75551ff4cf3e57da98979f6a937aaa9ddb3915bf68cc17d03df733be6646ed" dependencies = [ "anyhow", - "base64 0.22.1", + "base64 0.23.1", "bytes", "futures", - "http 1.3.1", - "http-body 1.0.1", + "http 1.5.0", "jiff", "log", "md-5 0.11.0", "mea", "percent-encoding", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-core", - "reqwest 0.13.4", "serde", "serde_json", "tokio", @@ -9650,22 +9659,36 @@ dependencies = [ ] [[package]] -name = "opendal-layer-concurrent-limit" -version = "0.57.0" +name = "opendal-http-transport-reqwest" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d6f81ba6960e3fae1882f253b114b21d7e444e1534f209c7737a79f6243eb6f" +checksum = "ad4d4f19c3ce01126a30611f8e544eaa217104a278c889ac17c9374fe4f9e4ef" +dependencies = [ + "bytes", + "futures", + "http 1.5.0", + "http-body 1.0.1", + "opendal-core", + "reqwest 0.13.4", +] + +[[package]] +name = "opendal-layer-concurrent-limit" +version = "0.58.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "249ac5b0aa5a7a6c3737342d10456067937f9c9a6f3f02544271f7908ab91081" dependencies = [ "futures", - "http 1.3.1", + "http 1.5.0", "mea", "opendal-core", ] [[package]] name = "opendal-layer-logging" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "58ada45c6d81d1aa4c9305d0c7d4bc317c59c85866a0908a2d75a7a978aa5ee2" +checksum = "5c75411ab00f77851ff086b686c1e9ca8175ac18c15afa2cb75b9036436cb06c" dependencies = [ "log", "opendal-core", @@ -9673,20 +9696,20 @@ dependencies = [ [[package]] name = "opendal-layer-observe-metrics-common" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "628b0228fdbd13c3d9d50eee4341f2eb82ca5b44991e4c68f07c84cc823e2d12" +checksum = "12dbc31b27ee8e5f658ae6c8514b8276059e5235e2f85138a42e6ed172324876" dependencies = [ "futures", - "http 1.3.1", + "http 1.5.0", "opendal-core", ] [[package]] name = "opendal-layer-prometheus" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0487bdb1357097ec8654781bad03ef310282517738e2864ebde69e27aaafc5ec" +checksum = "c14e6cb0b3ea8d83bef751613b8bf4512678afe8b49c2c96a633515bc0b34d79" dependencies = [ "opendal-core", "opendal-layer-observe-metrics-common", @@ -9695,9 +9718,9 @@ dependencies = [ [[package]] name = "opendal-layer-retry" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b2a25a718afb81fad81cb9a0580a1cb989221fa2317f888c6a37f8dad408eb7" +checksum = "80b7738bd5f233ad8da39af9b9316b9b7a4eaddd91e8e32a1e19b7030688121d" dependencies = [ "backon", "log", @@ -9706,9 +9729,9 @@ dependencies = [ [[package]] name = "opendal-layer-timeout" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e91f731724c213af81e9d03517859c8fc47b4578e64ad61ae4f099f10fe36e3" +checksum = "a704141924500f3803c05ed871b53305d2a2f11cb5ef20160c3ee688a1857f66" dependencies = [ "opendal-core", "tokio", @@ -9716,29 +9739,29 @@ dependencies = [ [[package]] name = "opendal-layer-tracing" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90c6fc9df6da1f0dafbdf55fa48525f1643aefbe7da8f46936e869e2a5b8a34f" +checksum = "4827cf8b2ebd9123b1fb7286c421be079948e2b963a452f1f7e8aee2c12e7997" dependencies = [ "futures", - "http 1.3.1", + "http 1.5.0", "opendal-core", "tracing", ] [[package]] name = "opendal-service-azblob" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0030644366ef5d8cbe3a4a5822bf99a4aafddc1666e9d24b44d158d9062fc76a" +checksum = "b3310fbbb48f111c6f590473c2cd15e1b7f8e384444b0d4e328f0464c864d767" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bytes", - "http 1.3.1", + "http 1.5.0", "log", "opendal-core", "opendal-service-azure-common", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-azure-storage", "reqsign-core", "reqsign-file-read-tokio", @@ -9749,19 +9772,19 @@ dependencies = [ [[package]] name = "opendal-service-azure-common" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b489f13c42e69d69bdd72952b634356ec43a7881a20259b38b540fcecdf4051" +checksum = "7348c88edf15af435b7be930077746b569fac5e738c1bf6a363b675e7317c9df" dependencies = [ - "http 1.3.1", + "http 1.5.0", "opendal-core", ] [[package]] name = "opendal-service-fs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22e89a665fef0e6bd249cf5ea47fc174b7ba892159bee4b9382528b1ca873a2c" +checksum = "826c4e17a30643b888fe983897f9a4b23b07066e1d069727a923cc8fb419a702" dependencies = [ "bytes", "log", @@ -9773,17 +9796,17 @@ dependencies = [ [[package]] name = "opendal-service-gcs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48de101aac565ed06af4b47903c24eafd249075553ec1fb18256751c45148d47" +checksum = "007f3fba63c21e516c956b891e96ff9892d8175662bfb781cdada9d3766a11e6" dependencies = [ "async-trait", "bytes", - "http 1.3.1", + "http 1.5.0", "log", "opendal-core", "percent-encoding", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-core", "reqsign-file-read-tokio", "reqsign-google", @@ -9794,11 +9817,11 @@ dependencies = [ [[package]] name = "opendal-service-http" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb6af628a0bf14075b957179444927e1df40dc7addef382b585a05ef015a077b" +checksum = "d2fd1f0f28bd052e068d9bb4bc5d30a09de22a4f7cea24558dbeef2b91090cc3" dependencies = [ - "http 1.3.1", + "http 1.5.0", "log", "opendal-core", "serde", @@ -9806,9 +9829,9 @@ dependencies = [ [[package]] name = "opendal-service-mysql" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5cb3364a16d32aeb9c9c3232a81648926a5a8efa680f3316e0743ca2d5a37744" +checksum = "2d4a25f582068ce30ba24eb3efd4ac16ca833c87b5282d8d79634fdbead74d84" dependencies = [ "mea", "opendal-core", @@ -9818,15 +9841,15 @@ dependencies = [ [[package]] name = "opendal-service-oss" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "328fa55e8888cbdfe00826bfea2a79042422b720e8369e9e021e46121dea5ace" +checksum = "cd528ec2d49c5ca69e674ffed7b3e0686fb9cfcfea0596870de381467fda4f1b" dependencies = [ "bytes", - "http 1.3.1", + "http 1.5.0", "log", "opendal-core", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-aliyun-oss", "reqsign-core", "reqsign-file-read-tokio", @@ -9835,18 +9858,18 @@ dependencies = [ [[package]] name = "opendal-service-s3" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "313d46c9f5ae70bca26b7c3e3fbb9b639292625f28af73aa016f47e788af9deb" +checksum = "58e80cdf192d7eff05feed747894d64f81905ac4eaf132edf7ea270abdd2d663" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bytes", - "crc32c", - "http 1.3.1", + "crc-fast", + "http 1.5.0", "log", "md-5 0.11.0", "opendal-core", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-aws-v4", "reqsign-core", "reqsign-file-read-tokio", @@ -9898,7 +9921,7 @@ checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" dependencies = [ "async-trait", "bytes", - "http 1.3.1", + "http 1.5.0", "opentelemetry", "reqwest 0.13.4", ] @@ -9909,7 +9932,7 @@ version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" dependencies = [ - "http 1.3.1", + "http 1.5.0", "opentelemetry", "opentelemetry-http", "opentelemetry-proto", @@ -11835,19 +11858,9 @@ dependencies = [ [[package]] name = "quick-xml" -version = "0.39.4" +version = "0.41.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e" -dependencies = [ - "memchr", - "serde", -] - -[[package]] -name = "quick-xml" -version = "0.40.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2474bd2e5029e7ccb6abb2ba48cf2383a333851dedf495901544281590c7da7f" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" dependencies = [ "memchr", "serde", @@ -12293,13 +12306,13 @@ checksum = "51743d3e274e2b18df81c4dc6caf8a5b8e15dbe799e0dca05c7617380094e884" [[package]] name = "reqsign-aliyun-oss" -version = "3.1.0" +version = "3.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "372266b4733756738eeb199a98188037d27a0989980e2600ae7ce1faf00a867d" +checksum = "a5e6d659fcdbca6fe2d7ef109c2e28499b7be80501f1bb86c10caf5ec8ac1219" dependencies = [ "anyhow", "form_urlencoded", - "http 1.3.1", + "http 1.5.0", "log", "percent-encoding", "reqsign-core", @@ -12309,19 +12322,18 @@ dependencies = [ ] [[package]] -name = "reqsign-aws-v4" -version = "3.0.1" +name = "reqsign-aws-core" +version = "3.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b75624bd8a466e37ddc0a7b6c33ac859a85347c153a916e1dd9d0b68338f74a" +checksum = "e4af084e1f3cbf3e67e0c972765399bce54ecec804cceba46b39a8331f3c1bff" dependencies = [ - "anyhow", "bytes", "form_urlencoded", "hex", - "http 1.3.1", + "http 1.5.0", "log", "percent-encoding", - "quick-xml 0.40.1", + "quick-xml 0.41.0", "reqsign-core", "rust-ini 0.21.1", "serde", @@ -12331,16 +12343,31 @@ dependencies = [ ] [[package]] -name = "reqsign-azure-storage" -version = "3.0.1" +name = "reqsign-aws-v4" +version = "3.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62b96928e73ad984de1d99e382749d09e5dab7dd707b767974f7e40aa926b82f" +checksum = "4ac5b3b7cefa28933792b439186459f77f19f9b6edbeab41b8b187150361a206" +dependencies = [ + "bytes", + "http 1.5.0", + "log", + "quick-xml 0.41.0", + "reqsign-aws-core", + "reqsign-core", + "serde", +] + +[[package]] +name = "reqsign-azure-storage" +version = "3.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2824e7da3c2cc42ac3406c674eb57c89127fdcd97f3a73c608cfc680505ea134" dependencies = [ "anyhow", - "base64 0.22.1", + "base64 0.23.1", "bytes", "form_urlencoded", - "http 1.3.1", + "http 1.5.0", "log", "pem", "percent-encoding", @@ -12353,18 +12380,17 @@ dependencies = [ [[package]] name = "reqsign-core" -version = "3.0.1" +version = "3.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5fa5cb48808693614d1701fcd3db0b30fa292e0f18e122ae068b6d32eaeed3f" +checksum = "c07dd510b1e1b9b241883e483358147fb2ed2d497a7b39b065ba61eb93deceb0" dependencies = [ "anyhow", - "base64 0.22.1", + "base64 0.23.1", "bytes", - "form_urlencoded", "futures", "hex", "hmac 0.13.0", - "http 1.3.1", + "http 1.5.0", "jiff", "log", "percent-encoding", @@ -12378,9 +12404,9 @@ dependencies = [ [[package]] name = "reqsign-file-read-tokio" -version = "3.0.1" +version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a4b6f3a3fd29ffcc99a90aec585a65217783badfd73acddf847b63ae683bda9" +checksum = "663d9d55abd0df0830ef0ae43708297cc1371cf4e8ca91f3ac813c309cca8c98" dependencies = [ "anyhow", "reqsign-core", @@ -12389,12 +12415,12 @@ dependencies = [ [[package]] name = "reqsign-google" -version = "3.0.1" +version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb215d0876a18b6bd9cdd380b589e5292aaa638ca15266de794b1122d898b6b2" +checksum = "4080a227f82a09f68540ecd028622065d7ac4c0bcb8727a25bdcfc0526235792" dependencies = [ "form_urlencoded", - "http 1.3.1", + "http 1.5.0", "log", "percent-encoding", "reqsign-aws-v4", @@ -12416,7 +12442,7 @@ dependencies = [ "futures-core", "futures-util", "h2 0.4.11", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "hyper 1.6.0", @@ -12459,7 +12485,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "hyper 1.6.0", @@ -12499,7 +12525,7 @@ checksum = "57f17d28a6e6acfe1733fe24bcd30774d13bffa4b8a22535b4c8c98423088d4e" dependencies = [ "anyhow", "async-trait", - "http 1.3.1", + "http 1.5.0", "reqwest 0.12.24", "serde", "thiserror 1.0.69", @@ -12516,7 +12542,7 @@ dependencies = [ "async-trait", "futures", "getrandom 0.2.16", - "http 1.3.1", + "http 1.5.0", "hyper 1.6.0", "reqwest 0.12.24", "reqwest-middleware", @@ -12969,7 +12995,7 @@ dependencies = [ "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -13492,7 +13518,7 @@ dependencies = [ "headers", "hex", "hostname 0.3.1", - "http 1.3.1", + "http 1.5.0", "humantime", "humantime-serde", "hyper 1.6.0", @@ -13883,7 +13909,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -13919,6 +13945,12 @@ dependencies = [ "lock_api", ] +[[package]] +name = "spin" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3" + [[package]] name = "spki" version = "0.7.3" @@ -14880,7 +14912,7 @@ dependencies = [ "getrandom 0.3.4", "once_cell", "rustix 1.0.7", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -15017,7 +15049,7 @@ dependencies = [ "futures", "futures-util", "hex", - "http 1.3.1", + "http 1.5.0", "hyper-util", "itertools 0.14.0", "jsonb", @@ -15522,7 +15554,7 @@ dependencies = [ "bytes", "flate2", "h2 0.4.11", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "hyper 1.6.0", @@ -15615,7 +15647,7 @@ checksum = "29453d84de05f4f1b573db22e6f9f6c95c189a6089a440c9a098aa9dea009299" dependencies = [ "base64 0.22.1", "bytes", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "pin-project", "tokio-stream", @@ -15674,7 +15706,7 @@ dependencies = [ "base64 0.21.7", "bitflags 2.12.1", "bytes", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "mime", @@ -15696,7 +15728,7 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http 1.3.1", + "http 1.5.0", "http-body 1.0.1", "http-body-util", "http-range-header", diff --git a/Cargo.toml b/Cargo.toml index cc46034f1b..77fd4dbade 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -181,7 +181,7 @@ nalgebra = "0.33" nix = { version = "0.30.1", default-features = false, features = ["event", "fs", "process"] } notify = "8.0" num_cpus = "1.16" -object_store_opendal = "0.57" +object_store_opendal = "0.58" once_cell = "1.18" opentelemetry-proto = { version = "0.32", features = [ "gen-tonic", diff --git a/src/cli/src/common/object_store.rs b/src/cli/src/common/object_store.rs index 129fb29cb7..e671e6a534 100644 --- a/src/cli/src/common/object_store.rs +++ b/src/cli/src/common/object_store.rs @@ -389,8 +389,7 @@ pub fn new_fs_object_store(root: &str) -> std::result::Result::from(&config)) .context(error::InitBackendSnafu) - .map_err(BoxedError::new)? - .finish(); + .map_err(BoxedError::new)?; Ok(with_instrument_layers( with_retry_layers(object_store), false, diff --git a/src/cli/src/data/snapshot_storage.rs b/src/cli/src/data/snapshot_storage.rs index 715ae79dec..17da8aea0b 100644 --- a/src/cli/src/data/snapshot_storage.rs +++ b/src/cli/src/data/snapshot_storage.rs @@ -338,9 +338,7 @@ impl OpenDalStorage { let path = extract_file_path_from_uri(uri)?; let builder = Fs::default().root(&path); - let object_store = ObjectStore::new(builder) - .context(BuildObjectStoreSnafu)? - .finish(); + let object_store = ObjectStore::new(builder).context(BuildObjectStoreSnafu)?; Ok(Self::new_operator_rooted( Self::finish_local_store(object_store), uri, @@ -381,9 +379,7 @@ impl OpenDalStorage { Self::validate_remote_config(uri, "s3", config.validate())?; let conn: S3Connection = config.into(); - let object_store = ObjectStore::new(S3::from(&conn)) - .context(BuildObjectStoreSnafu)? - .finish(); + let object_store = ObjectStore::new(S3::from(&conn)).context(BuildObjectStoreSnafu)?; Ok(Self::new_operator_rooted( Self::finish_remote_store(object_store), uri, @@ -412,9 +408,7 @@ impl OpenDalStorage { Self::validate_remote_config(uri, "oss", config.validate())?; let conn: OssConnection = config.into(); - let object_store = ObjectStore::new(Oss::from(&conn)) - .context(BuildObjectStoreSnafu)? - .finish(); + let object_store = ObjectStore::new(Oss::from(&conn)).context(BuildObjectStoreSnafu)?; Ok(Self::new_operator_rooted( Self::finish_remote_store(object_store), uri, @@ -448,9 +442,7 @@ impl OpenDalStorage { } let conn: GcsConnection = config.into(); - let object_store = ObjectStore::new(Gcs::from(&conn)) - .context(BuildObjectStoreSnafu)? - .finish(); + let object_store = ObjectStore::new(Gcs::from(&conn)).context(BuildObjectStoreSnafu)?; Ok(Self::new_operator_rooted( Self::finish_remote_store(object_store), uri, @@ -500,9 +492,7 @@ impl OpenDalStorage { Self::validate_remote_config(uri, "azblob", config.validate())?; let conn: AzblobConnection = config.into(); - let object_store = ObjectStore::new(Azblob::from(&conn)) - .context(BuildObjectStoreSnafu)? - .finish(); + let object_store = ObjectStore::new(Azblob::from(&conn)).context(BuildObjectStoreSnafu)?; Ok(Self::new_operator_rooted( Self::finish_remote_store(object_store), uri, @@ -732,9 +722,7 @@ mod tests { use crate::data::export_v2::schema::SchemaDefinition; fn make_storage_with_rooted_fs(dir: &std::path::Path) -> OpenDalStorage { - let object_store = ObjectStore::new(Fs::default().root(dir.to_str().unwrap())) - .unwrap() - .finish(); + let object_store = ObjectStore::new(Fs::default().root(dir.to_str().unwrap())).unwrap(); OpenDalStorage::new_operator_rooted( OpenDalStorage::finish_local_store(object_store), Url::from_directory_path(dir).unwrap().as_ref(), diff --git a/src/cmd/src/bin/query_perf_fixture/direct_sst.rs b/src/cmd/src/bin/query_perf_fixture/direct_sst.rs index 7297a54b2c..41b70a9690 100644 --- a/src/cmd/src/bin/query_perf_fixture/direct_sst.rs +++ b/src/cmd/src/bin/query_perf_fixture/direct_sst.rs @@ -441,8 +441,7 @@ pub(super) async fn run_direct_sst(args: DirectArgs) { fs::create_dir_all(&obj_store_dir).expect("failed to create fixture object-store directory"); fs::create_dir_all(&manifest_dir).expect("failed to create fixture manifest directory"); let ostorage = ObjectStore::new(FsBuilder::default().root(&obj_store_dir.to_string_lossy())) - .expect("failed to create filesystem object store for fixture output") - .finish(); + .expect("failed to create filesystem object store for fixture output"); let metadata: RegionMetadataRef = Arc::new(build_region_metadata(table, region_id)); let mut files = HashMap::with_capacity(scenario.layout.sst_count); let mut next_file_index = 1; diff --git a/src/cmd/src/bin/query_perf_fixture/inspect_footer.rs b/src/cmd/src/bin/query_perf_fixture/inspect_footer.rs index 4bc9845671..d2a2e8006c 100644 --- a/src/cmd/src/bin/query_perf_fixture/inspect_footer.rs +++ b/src/cmd/src/bin/query_perf_fixture/inspect_footer.rs @@ -156,13 +156,13 @@ async fn build_store( let operator = match destination { Some(destination) => match &destination.object_store { ObjectStoreConfig::File(_) => { - object_store::ObjectStore::new(Fs::default().root(&destination.data_home))?.finish() + object_store::ObjectStore::new(Fs::default().root(&destination.data_home))? } _ => new_raw_object_store(&destination.object_store, &destination.data_home).await?, }, None => { let root = root.expect("root must be set when destination is None"); - object_store::ObjectStore::new(Fs::default().root(&root.to_string_lossy()))?.finish() + object_store::ObjectStore::new(Fs::default().root(&root.to_string_lossy()))? } }; Ok(Arc::new(object_store_opendal::OpendalStore::new(operator))) diff --git a/src/cmd/src/bin/query_regression_runner/materialize.rs b/src/cmd/src/bin/query_regression_runner/materialize.rs index 3cc9e1aa07..f35d4d7241 100644 --- a/src/cmd/src/bin/query_regression_runner/materialize.rs +++ b/src/cmd/src/bin/query_regression_runner/materialize.rs @@ -66,7 +66,9 @@ async fn materialize( } fn fs_operator(root: &Path) -> Result { - Ok(ObjectStore::new(Fs::default().root(&root.to_string_lossy()))?.finish()) + Ok(ObjectStore::new( + Fs::default().root(&root.to_string_lossy()), + )?) } fn validate_region_dir(region_dir: &str) -> Result { diff --git a/src/cmd/src/datanode/parquetbench.rs b/src/cmd/src/datanode/parquetbench.rs index 823be0e188..a5c73f2952 100644 --- a/src/cmd/src/datanode/parquetbench.rs +++ b/src/cmd/src/datanode/parquetbench.rs @@ -622,8 +622,7 @@ fn build_local_file_source(file_path: &Path) -> error::Result Vec { pub fn test_store(root: &str) -> ObjectStore { let builder = Fs::default(); - ObjectStore::new(builder.root(root)).unwrap().finish() + ObjectStore::new(builder.root(root)).unwrap() } pub fn test_tmp_store(root: &str) -> (ObjectStore, TempDir) { let dir = create_temp_dir(root); let builder = Fs::default(); - (ObjectStore::new(builder.root("/")).unwrap().finish(), dir) + (ObjectStore::new(builder.root("/")).unwrap(), dir) } pub fn test_basic_schema() -> SchemaRef { diff --git a/src/common/meta/src/snapshot.rs b/src/common/meta/src/snapshot.rs index e66156b255..f35c86e2e7 100644 --- a/src/common/meta/src/snapshot.rs +++ b/src/common/meta/src/snapshot.rs @@ -399,7 +399,7 @@ mod tests { let temp_path = temp_dir.path(); let data_path = temp_path.join("data").as_path().display().to_string(); let builder = Fs::default().root(&data_path); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let manager = MetadataSnapshotManager::new(kv_backend.clone(), object_store); (temp_dir, kv_backend, manager) } diff --git a/src/common/procedure/src/local.rs b/src/common/procedure/src/local.rs index c870cc39a6..2ae7f5e3e0 100644 --- a/src/common/procedure/src/local.rs +++ b/src/common/procedure/src/local.rs @@ -1039,7 +1039,7 @@ pub(crate) mod test_util { pub(crate) fn new_object_store(dir: &TempDir) -> ObjectStore { let store_dir = dir.path().to_str().unwrap(); let builder = Builder::default(); - ObjectStore::new(builder.root(store_dir)).unwrap().finish() + ObjectStore::new(builder.root(store_dir)).unwrap() } } diff --git a/src/common/procedure/src/store.rs b/src/common/procedure/src/store.rs index b2f19b2f94..b97dab00d0 100644 --- a/src/common/procedure/src/store.rs +++ b/src/common/procedure/src/store.rs @@ -391,7 +391,7 @@ mod tests { fn procedure_store_for_test(dir: &TempDir) -> ProcedureStore { let store_dir = dir.path().to_str().unwrap(); let builder = Builder::default().root(store_dir); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); ProcedureStore::from_object_store(object_store) } diff --git a/src/common/procedure/src/store/state_store.rs b/src/common/procedure/src/store/state_store.rs index f4785becb2..7c36312d60 100644 --- a/src/common/procedure/src/store/state_store.rs +++ b/src/common/procedure/src/store/state_store.rs @@ -223,7 +223,7 @@ mod tests { let store_dir = dir.path().to_str().unwrap(); let builder = Builder::default().root(store_dir); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let state_store = ObjectStateStore::new(object_store); let data: Vec<_> = state_store @@ -293,7 +293,7 @@ mod tests { let store_dir = dir.path().to_str().unwrap(); let builder = Builder::default().root(store_dir); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let state_store = ObjectStateStore::new(object_store); state_store.put("a/1", b"v1".to_vec()).await.unwrap(); diff --git a/src/file-engine/src/engine.rs b/src/file-engine/src/engine.rs index fae7a1f2f0..05be293770 100644 --- a/src/file-engine/src/engine.rs +++ b/src/file-engine/src/engine.rs @@ -396,9 +396,7 @@ mod tests { use crate::error::Error; fn build_fs_object_store() -> ObjectStore { - ObjectStore::new(Fs::default().root("/tmp")) - .unwrap() - .finish() + ObjectStore::new(Fs::default().root("/tmp")).unwrap() } fn build_s3_object_store() -> ObjectStore { @@ -409,7 +407,6 @@ mod tests { .disable_ec2_metadata(), ) .unwrap() - .finish() } #[test] diff --git a/src/file-engine/src/test_util.rs b/src/file-engine/src/test_util.rs index b3b4d69cfb..e97396bb4f 100644 --- a/src/file-engine/src/test_util.rs +++ b/src/file-engine/src/test_util.rs @@ -27,7 +27,7 @@ pub fn new_test_object_store(prefix: &str) -> (TempDir, ObjectStore) { let dir = create_temp_dir(prefix); let store_dir = dir.path().to_string_lossy(); let builder = Fs::default().root(&store_dir); - (dir, ObjectStore::new(builder).unwrap().finish()) + (dir, ObjectStore::new(builder).unwrap()) } pub fn new_test_column_metadata() -> Vec { diff --git a/src/log-store/src/kafka/index/collector.rs b/src/log-store/src/kafka/index/collector.rs index 8fd1fba39e..54b9b21553 100644 --- a/src/log-store/src/kafka/index/collector.rs +++ b/src/log-store/src/kafka/index/collector.rs @@ -339,9 +339,8 @@ mod tests { #[tokio::test] async fn test_read_remote_region_index() { - let operator = object_store::ObjectStore::new(object_store::services::Memory::default()) - .unwrap() - .finish(); + let operator = + object_store::ObjectStore::new(object_store::services::Memory::default()).unwrap(); let path = default_index_file(0); let encoder = JsonIndexEncoder::default(); diff --git a/src/metric-engine/src/test_util.rs b/src/metric-engine/src/test_util.rs index 71e5495eca..1f17737098 100644 --- a/src/metric-engine/src/test_util.rs +++ b/src/metric-engine/src/test_util.rs @@ -492,7 +492,7 @@ mod test { let region_id = to_metadata_region_id(env.default_physical_region_id()); let builder = Fs::default().root(&env.data_home()); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let table_dir = TestEnv::default_table_dir(); let region_dir = join_dir(&table_dir, "1_0000000002"); diff --git a/src/mito2/src/access_layer.rs b/src/mito2/src/access_layer.rs index 79535604d8..8580884274 100644 --- a/src/mito2/src/access_layer.rs +++ b/src/mito2/src/access_layer.rs @@ -663,7 +663,7 @@ pub(crate) async fn new_fs_cache_store(root: &str) -> Result { clean_dir(&old_atomic_temp_dir).await?; let builder = Fs::default().root(root).atomic_write_dir(&atomic_write_dir); - let store = ObjectStore::new(builder).context(OpenDalSnafu)?.finish(); + let store = ObjectStore::new(builder).context(OpenDalSnafu)?; Ok(with_instrument_layers(store, false)) } diff --git a/src/mito2/src/cache/file_cache.rs b/src/mito2/src/cache/file_cache.rs index 64f2fad6f1..0594a1ac10 100644 --- a/src/mito2/src/cache/file_cache.rs +++ b/src/mito2/src/cache/file_cache.rs @@ -858,7 +858,7 @@ mod tests { fn new_fs_store(path: &str) -> ObjectStore { let builder = Fs::default().root(path); - ObjectStore::new(builder).unwrap().finish() + ObjectStore::new(builder).unwrap() } #[tokio::test] diff --git a/src/mito2/src/cache/manifest_cache.rs b/src/mito2/src/cache/manifest_cache.rs index 22fd796fac..f57cc3c06d 100644 --- a/src/mito2/src/cache/manifest_cache.rs +++ b/src/mito2/src/cache/manifest_cache.rs @@ -481,7 +481,7 @@ mod tests { fn new_fs_store(path: &str) -> ObjectStore { let builder = Fs::default().root(path); - ObjectStore::new(builder).unwrap().finish() + ObjectStore::new(builder).unwrap() } #[tokio::test] diff --git a/src/mito2/src/cache/test_util.rs b/src/mito2/src/cache/test_util.rs index c5baae3c75..58b8272fbc 100644 --- a/src/mito2/src/cache/test_util.rs +++ b/src/mito2/src/cache/test_util.rs @@ -83,7 +83,7 @@ fn parquet_file_data_inner(key_value_metadata: Option>) -> Vec pub(crate) fn new_fs_store(path: &str) -> ObjectStore { let builder = Fs::default(); - ObjectStore::new(builder.root(path)).unwrap().finish() + ObjectStore::new(builder.root(path)).unwrap() } pub(crate) fn assert_parquet_metadata_equal(x: Arc, y: Arc) { diff --git a/src/mito2/src/cache/write_cache.rs b/src/mito2/src/cache/write_cache.rs index b464e366cb..03c7bdddd7 100644 --- a/src/mito2/src/cache/write_cache.rs +++ b/src/mito2/src/cache/write_cache.rs @@ -587,9 +587,9 @@ mod tests { #[tokio::test] async fn test_upload_uses_wrapped_remote_store() { let env = TestEnv::new().await; - let local_store = ObjectStore::new(Memory::default()).unwrap().finish(); - let original_store = ObjectStore::new(Memory::default()).unwrap().finish(); - let target_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let local_store = ObjectStore::new(Memory::default()).unwrap(); + let original_store = ObjectStore::new(Memory::default()).unwrap(); + let target_store = ObjectStore::new(Memory::default()).unwrap(); let wrapper = Arc::new(RedirectUploadStoreWrapper { target_store: target_store.clone(), last_op_type: Mutex::new(None), diff --git a/src/mito2/src/gc/worker_test.rs b/src/mito2/src/gc/worker_test.rs index e0b0348499..7bcf511f1a 100644 --- a/src/mito2/src/gc/worker_test.rs +++ b/src/mito2/src/gc/worker_test.rs @@ -556,7 +556,7 @@ async fn write_and_list_entry(store: &ObjectStore, path: &str) -> Entry { #[tokio::test] async fn test_unknown_file_within_ttl_not_deleted() { let builder = services::Memory::default(); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "test/1.parquet").await; @@ -586,7 +586,7 @@ async fn test_unknown_file_exceeded_ttl_deleted() { let tmp_dir = common_test_util::temp_dir::create_temp_dir("gc_unknown_ttl"); let root = tmp_dir.path().to_string_lossy().to_string(); let builder = services::Fs::default().root(&root); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "2.parquet").await; @@ -612,7 +612,7 @@ async fn test_unknown_file_exceeded_ttl_deleted() { #[tokio::test] async fn test_unknown_file_dropped_region_deleted() { let builder = services::Memory::default(); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "test/3.parquet").await; @@ -637,7 +637,7 @@ async fn test_unknown_file_dropped_region_deleted() { #[tokio::test] async fn test_file_in_manifest_not_deleted() { let builder = services::Memory::default(); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "test/4.parquet").await; let threshold = chrono::Utc::now() + chrono::Duration::days(1); @@ -655,7 +655,7 @@ async fn test_file_in_manifest_not_deleted() { #[tokio::test] async fn test_file_in_tmp_ref_not_deleted() { let builder = services::Memory::default(); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "test/5.parquet").await; let threshold = chrono::Utc::now() + chrono::Duration::days(1); @@ -673,7 +673,7 @@ async fn test_file_in_tmp_ref_not_deleted() { #[tokio::test] async fn test_known_file_still_lingering_not_deleted() { let builder = services::Memory::default(); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "test/6.parquet").await; let threshold = chrono::Utc::now() + chrono::Duration::days(1); @@ -694,7 +694,7 @@ async fn test_known_file_still_lingering_not_deleted() { #[tokio::test] async fn test_known_file_eligible_for_delete_deleted() { let builder = services::Memory::default(); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "test/7.parquet").await; let threshold = chrono::DateTime::from_timestamp(0, 0).unwrap(); @@ -720,7 +720,7 @@ async fn test_unknown_file_at_cutoff_not_deleted() { let tmp_dir = common_test_util::temp_dir::create_temp_dir("gc_at_cutoff"); let root = tmp_dir.path().to_string_lossy().to_string(); let builder = services::Fs::default().root(&root); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "cutoff.parquet").await; @@ -753,7 +753,7 @@ async fn test_unknown_file_at_cutoff_not_deleted() { async fn test_missing_last_modified_unknown_kept() { // Memory backend does NOT set last_modified let builder = services::Memory::default(); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "test/missing_mtime.parquet").await; // Verify that last_modified is indeed None @@ -783,7 +783,7 @@ async fn test_file_in_manifest_old_mtime_kept() { let tmp_dir = common_test_util::temp_dir::create_temp_dir("gc_manifest_old"); let root = tmp_dir.path().to_string_lossy().to_string(); let builder = services::Fs::default().root(&root); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "in_manifest.parquet").await; // threshold far in the future → mtime is definitely old, but manifest protects @@ -808,7 +808,7 @@ async fn test_file_in_tmp_ref_old_mtime_kept() { let tmp_dir = common_test_util::temp_dir::create_temp_dir("gc_tmpref_old"); let root = tmp_dir.path().to_string_lossy().to_string(); let builder = services::Fs::default().root(&root); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let entry = write_and_list_entry(&store, "in_tmp_ref.parquet").await; // threshold far in the future → mtime is definitely old, but tmp_ref protects diff --git a/src/mito2/src/manifest/storage.rs b/src/mito2/src/manifest/storage.rs index 5f795cc1c9..9ebeef2e9a 100644 --- a/src/mito2/src/manifest/storage.rs +++ b/src/mito2/src/manifest/storage.rs @@ -486,7 +486,7 @@ mod tests { common_telemetry::init_default_ut_logging(); let tmp_dir = create_temp_dir("test_manifest_log_store"); let builder = Fs::default().root(&tmp_dir.path().to_string_lossy()); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); ManifestObjectStore::new( path, object_store, diff --git a/src/mito2/src/manifest/storage/staging.rs b/src/mito2/src/manifest/storage/staging.rs index dd0105a091..a0b350a76c 100644 --- a/src/mito2/src/manifest/storage/staging.rs +++ b/src/mito2/src/manifest/storage/staging.rs @@ -258,7 +258,7 @@ mod tests { let tmp_dir = create_temp_dir("test_staging_storage_clear"); let builder = Fs::default().root(&tmp_dir.path().to_string_lossy()); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let manifest_path = "/data/table/region_0001/manifest/"; let mut storage = StagingStorage::new( manifest_path.to_string(), diff --git a/src/mito2/src/manifest/tests/checkpoint.rs b/src/mito2/src/manifest/tests/checkpoint.rs index 86679630ac..0fe77502e1 100644 --- a/src/mito2/src/manifest/tests/checkpoint.rs +++ b/src/mito2/src/manifest/tests/checkpoint.rs @@ -21,8 +21,8 @@ use common_datasource::compression::CompressionType; use common_error::ext::{ErrorExt, RetryHint}; use common_error::status_code::StatusCode; use object_store::layers::mock::{ - Buffer, Error as MockError, ErrorKind, MockLayer, MockLayerBuilder, OpDelete, - Result as MockResult, oio, + BytesRange, Error as MockError, ErrorKind, MockLayer, MockLayerBuilder, OpDelete, + Result as MockResult, RpRead, oio, }; use store_api::storage::{FileId, RegionId}; use strum::IntoEnumIterator; @@ -90,8 +90,8 @@ fn nop_action() -> RegionMetaActionList { struct NotFoundReader; -impl oio::Read for NotFoundReader { - async fn read(&mut self) -> MockResult { +impl oio::StreamRead for NotFoundReader { + async fn open(&self, _: BytesRange) -> MockResult<(RpRead, Box)> { Err(MockError::new( ErrorKind::NotFound, "mock listed manifest delta not found", @@ -104,7 +104,7 @@ fn fail_manifest_delta_reads_layer() -> MockLayer { .reader_factory(Arc::new(|path, _args, inner| { let file_name = path.rsplit('/').next().unwrap_or(path); if is_delta_file(file_name) { - Box::new(NotFoundReader) + Box::new(oio::StreamReader::new(NotFoundReader)) } else { inner } diff --git a/src/mito2/src/region.rs b/src/mito2/src/region.rs index b68b0d1ccc..072ec6ca62 100644 --- a/src/mito2/src/region.rs +++ b/src/mito2/src/region.rs @@ -2499,7 +2499,7 @@ mod tests { let temp_dir = create_temp_dir(""); let path_str = temp_dir.path().display().to_string(); let fs_builder = Fs::default().root(&path_str); - let object_store = ObjectStore::new(fs_builder).unwrap().finish(); + let object_store = ObjectStore::new(fs_builder).unwrap(); let index_aux_path = temp_dir.path().join("index_aux"); let puffin_mgr = PuffinManagerFactory::new(&index_aux_path, 4096, None, None) diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs index ab2e0c737d..7e02c4945c 100644 --- a/src/mito2/src/region/opener.rs +++ b/src/mito2/src/region/opener.rs @@ -1413,9 +1413,7 @@ mod tests { } fn build_fs_object_store() -> ObjectStore { - ObjectStore::new(Fs::default().root("/tmp")) - .unwrap() - .finish() + ObjectStore::new(Fs::default().root("/tmp")).unwrap() } #[test] @@ -1497,8 +1495,7 @@ mod tests { .region("us-east-1") .disable_ec2_metadata(), ) - .unwrap() - .finish(); + .unwrap(); assert!(supports_open_region_object_storage_requirement( &object_store @@ -1553,7 +1550,7 @@ mod tests { async fn test_preload_parquet_meta_cache_uses_file_cache() { let env = TestEnv::new().await; - let local_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let local_store = ObjectStore::new(Memory::default()).unwrap(); let write_cache = env .create_write_cache(local_store, ReadableSize::mb(1024)) .await; @@ -1604,7 +1601,7 @@ mod tests { let path_type = PathType::Bare; let remote_path = file_handle.file_path(table_dir, path_type); - let source_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let source_store = ObjectStore::new(Memory::default()).unwrap(); source_store .write(&remote_path, parquet_bytes) .await @@ -1688,7 +1685,7 @@ mod tests { let remote_path = file_handle.file_path(table_dir, path_type); // Even if the remote object store has the file, we should not preload from it. - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); object_store .write(&remote_path, b"noop".as_slice()) .await @@ -1763,9 +1760,8 @@ mod tests { let file_path = file_handle.file_path(table_dir, path_type); let root = create_temp_dir("parquet-meta-preload"); - let object_store = ObjectStore::new(Fs::default().root(root.path().to_str().unwrap())) - .unwrap() - .finish(); + let object_store = + ObjectStore::new(Fs::default().root(root.path().to_str().unwrap())).unwrap(); object_store.write(&file_path, parquet_bytes).await.unwrap(); let region_file_id = file_handle.file_id(); diff --git a/src/mito2/src/series_index/searcher.rs b/src/mito2/src/series_index/searcher.rs index e923de9f64..87bc4eac9d 100644 --- a/src/mito2/src/series_index/searcher.rs +++ b/src/mito2/src/series_index/searcher.rs @@ -259,7 +259,7 @@ mod tests { use crate::test_util::sst_util::{new_sparse_primary_key, sst_region_metadata_with_encoding}; fn object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } fn flat_batch(primary_keys: &[Vec], timestamps: &[i64]) -> RecordBatch { diff --git a/src/mito2/src/series_index/writer.rs b/src/mito2/src/series_index/writer.rs index d297800c4d..5c0dc3be89 100644 --- a/src/mito2/src/series_index/writer.rs +++ b/src/mito2/src/series_index/writer.rs @@ -650,7 +650,7 @@ mod tests { use crate::test_util::sst_util::{new_sparse_primary_key, sst_region_metadata_with_encoding}; fn object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } fn flat_schema(primary_key_type: DataType) -> SchemaRef { diff --git a/src/mito2/src/sst/file_purger.rs b/src/mito2/src/sst/file_purger.rs index 70c14915af..5f0617751d 100644 --- a/src/mito2/src/sst/file_purger.rs +++ b/src/mito2/src/sst/file_purger.rs @@ -243,7 +243,7 @@ mod tests { .await .unwrap(); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let layer = Arc::new(AccessLayer::new( sst_dir, @@ -309,7 +309,7 @@ mod tests { .await .unwrap(); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let layer = Arc::new(AccessLayer::new( sst_dir, diff --git a/src/mito2/src/sst/index.rs b/src/mito2/src/sst/index.rs index 6b26cf5255..268b838739 100644 --- a/src/mito2/src/sst/index.rs +++ b/src/mito2/src/sst/index.rs @@ -1660,7 +1660,7 @@ mod tests { } fn mock_object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } async fn mock_intm_mgr(path: impl AsRef) -> IntermediateManager { diff --git a/src/mito2/src/sst/index/bloom_filter/applier.rs b/src/mito2/src/sst/index/bloom_filter/applier.rs index b1f6032630..e2d25e4bf8 100644 --- a/src/mito2/src/sst/index/bloom_filter/applier.rs +++ b/src/mito2/src/sst/index/bloom_filter/applier.rs @@ -517,7 +517,7 @@ mod tests { async fn test_compatible_predicate_for_sst() { let (_d, puffin_manager_factory) = PuffinManagerFactory::new_for_test_async("test_plan_for_sst_basic_").await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let table_dir = "table_dir".to_string(); let predicates = BTreeMap::from_iter([( @@ -545,7 +545,7 @@ mod tests { async fn test_compatible_predicate_for_sst_type_mismatch() { let (_d, puffin_manager_factory) = PuffinManagerFactory::new_for_test_async("test_plan_for_sst_type_mismatch_").await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let table_dir = "table_dir".to_string(); let predicates = BTreeMap::from_iter([( @@ -573,7 +573,7 @@ mod tests { async fn test_compatible_predicate_for_sst_partial_type_mismatch() { let (_d, puffin_manager_factory) = PuffinManagerFactory::new_for_test_async("test_plan_for_sst_partial_mismatch_").await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let table_dir = "table_dir".to_string(); // Column 1 (tag_str): expected string — matches SST (compatible). diff --git a/src/mito2/src/sst/index/bloom_filter/applier/builder.rs b/src/mito2/src/sst/index/bloom_filter/applier/builder.rs index e5968b5abb..4b8a11b097 100644 --- a/src/mito2/src/sst/index/bloom_filter/applier/builder.rs +++ b/src/mito2/src/sst/index/bloom_filter/applier/builder.rs @@ -375,7 +375,7 @@ mod tests { } fn test_object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } fn column(name: &str) -> Expr { diff --git a/src/mito2/src/sst/index/bloom_filter/creator.rs b/src/mito2/src/sst/index/bloom_filter/creator.rs index a7283f9191..7828cdb9fd 100644 --- a/src/mito2/src/sst/index/bloom_filter/creator.rs +++ b/src/mito2/src/sst/index/bloom_filter/creator.rs @@ -485,7 +485,7 @@ pub(crate) mod tests { use crate::sst::index::puffin_manager::PuffinManagerFactory; pub fn mock_object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } pub async fn new_intm_mgr(path: impl AsRef) -> IntermediateManager { diff --git a/src/mito2/src/sst/index/fulltext_index/creator.rs b/src/mito2/src/sst/index/fulltext_index/creator.rs index 58c3f1a9bc..800e0b77a0 100644 --- a/src/mito2/src/sst/index/fulltext_index/creator.rs +++ b/src/mito2/src/sst/index/fulltext_index/creator.rs @@ -489,7 +489,7 @@ mod tests { use crate::sst::index::puffin_manager::PuffinManagerFactory; fn mock_object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } async fn new_intm_mgr(path: impl AsRef) -> IntermediateManager { diff --git a/src/mito2/src/sst/index/inverted_index/applier.rs b/src/mito2/src/sst/index/inverted_index/applier.rs index a75962ec66..0f665eec54 100644 --- a/src/mito2/src/sst/index/inverted_index/applier.rs +++ b/src/mito2/src/sst/index/inverted_index/applier.rs @@ -436,7 +436,7 @@ mod tests { async fn test_plan_for_sst() { let (_d, puffin_manager_factory) = PuffinManagerFactory::new_for_test_async("test_plan_for_sst_basic_").await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let table_dir = "table_dir".to_string(); let mut predicates = BTreeMap::new(); @@ -468,7 +468,7 @@ mod tests { async fn test_plan_for_sst_type_mismatch() { let (_d, puffin_manager_factory) = PuffinManagerFactory::new_for_test_async("test_plan_for_sst_type_mismatch_").await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let table_dir = "table_dir".to_string(); let mut predicates = BTreeMap::new(); @@ -502,7 +502,7 @@ mod tests { let (_d, puffin_manager_factory) = PuffinManagerFactory::new_for_test_async("test_index_applier_apply_invalid_blob_type_") .await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let file_id = RegionFileId::new(0.into(), FileId::random()); let index_id = RegionIndexId::new(file_id, 0); let table_dir = "table_dir".to_string(); diff --git a/src/mito2/src/sst/index/inverted_index/applier/builder.rs b/src/mito2/src/sst/index/inverted_index/applier/builder.rs index 4938352792..d2d4c544bc 100644 --- a/src/mito2/src/sst/index/inverted_index/applier/builder.rs +++ b/src/mito2/src/sst/index/inverted_index/applier/builder.rs @@ -300,7 +300,7 @@ mod tests { } pub(crate) fn test_object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } pub(crate) fn tag_column() -> Expr { diff --git a/src/mito2/src/sst/index/inverted_index/creator.rs b/src/mito2/src/sst/index/inverted_index/creator.rs index 952303f33f..9cad6fa84e 100644 --- a/src/mito2/src/sst/index/inverted_index/creator.rs +++ b/src/mito2/src/sst/index/inverted_index/creator.rs @@ -471,7 +471,7 @@ mod tests { use crate::sst::index::puffin_manager::PuffinManagerFactory; fn mock_object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } async fn new_intm_mgr(path: impl AsRef) -> IntermediateManager { diff --git a/src/mito2/src/sst/index/puffin_manager.rs b/src/mito2/src/sst/index/puffin_manager.rs index edff8aab58..26471b4ea3 100644 --- a/src/mito2/src/sst/index/puffin_manager.rs +++ b/src/mito2/src/sst/index/puffin_manager.rs @@ -215,7 +215,7 @@ mod tests { let (_dir, factory) = PuffinManagerFactory::new_for_test_async("test_puffin_manager_factory_").await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let manager = factory.build(object_store, TestFilePathProvider); let file_id = RegionIndexId::new(RegionFileId::new(0.into(), FileId::random()), 0); diff --git a/src/mito2/src/sst/index/store.rs b/src/mito2/src/sst/index/store.rs index 62a5981298..e737252652 100644 --- a/src/mito2/src/sst/index/store.rs +++ b/src/mito2/src/sst/index/store.rs @@ -358,7 +358,7 @@ mod tests { #[tokio::test] async fn test_instrumented_store_read_write() { let instrumented_store = - InstrumentedStore::new(ObjectStore::new(Memory::default()).unwrap().finish()); + InstrumentedStore::new(ObjectStore::new(Memory::default()).unwrap()); let read_byte_count = IntCounter::new("read_byte_count", "read_byte_count").unwrap(); let read_count = IntCounter::new("read_count", "read_count").unwrap(); diff --git a/src/mito2/src/sst/index/vector_index/applier.rs b/src/mito2/src/sst/index/vector_index/applier.rs index dcdd044f19..40c8c25727 100644 --- a/src/mito2/src/sst/index/vector_index/applier.rs +++ b/src/mito2/src/sst/index/vector_index/applier.rs @@ -356,7 +356,7 @@ mod tests { ) -> (TempDir, VectorIndexApplier, RegionIndexId, u64) { let (dir, puffin_manager_factory) = PuffinManagerFactory::new_for_test_async("test_vector_index_applier_").await; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let file_id = RegionFileId::new(0.into(), FileId::random()); let index_id = RegionIndexId::new(file_id, 0); let table_dir = "table_dir".to_string(); diff --git a/src/mito2/src/sst/parquet/metadata.rs b/src/mito2/src/sst/parquet/metadata.rs index 9d38d6e216..3c0d197c5c 100644 --- a/src/mito2/src/sst/parquet/metadata.rs +++ b/src/mito2/src/sst/parquet/metadata.rs @@ -283,7 +283,7 @@ mod tests { let parquet_bytes = build_test_parquet_bytes(false, &[], &[4], EnabledStatistics::Page); let file_size = parquet_bytes.len() as u64; let file_path = "test.parquet"; - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); object_store.write(file_path, parquet_bytes).await.unwrap(); let mut loader = MetadataLoader::new(object_store, file_path, file_size); diff --git a/src/mito2/src/sst/parquet/reader.rs b/src/mito2/src/sst/parquet/reader.rs index 4fb98efeca..ff58d920b1 100644 --- a/src/mito2/src/sst/parquet/reader.rs +++ b/src/mito2/src/sst/parquet/reader.rs @@ -2707,9 +2707,7 @@ mod tests { // Persist the complete nested schema to an in-memory Parquet file so the projection is // exercised through parquet-rs rather than a mock. - let object_store = ObjectStore::new(Memory::default()) - .map_err(|e| e.to_string())? - .finish(); + let object_store = ObjectStore::new(Memory::default()).map_err(|e| e.to_string())?; let file_handle = sst_file_handle(0, 1); let file_path = file_handle.file_path("test_table", PathType::Bare); @@ -2824,7 +2822,7 @@ mod tests { } } - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let file_handle = sst_file_handle(0, 1); let table_dir = "test_table".to_string(); let path_type = PathType::Bare; @@ -2952,7 +2950,7 @@ mod tests { #[tokio::test(flavor = "current_thread")] async fn test_has_row_level_selection() { - let object_store = ObjectStore::new(Memory::default()).unwrap().finish(); + let object_store = ObjectStore::new(Memory::default()).unwrap(); let file_path = "row_level_selection.parquet"; let col = Arc::new(Int64Array::from_iter_values([1, 2, 3, 4, 5])) as ArrayRef; diff --git a/src/mito2/src/sst/range_index/searcher.rs b/src/mito2/src/sst/range_index/searcher.rs index b86c68c887..cc9c1b939a 100644 --- a/src/mito2/src/sst/range_index/searcher.rs +++ b/src/mito2/src/sst/range_index/searcher.rs @@ -323,7 +323,7 @@ mod tests { use crate::test_util::sst_util::{new_sparse_primary_key, sst_region_metadata_with_encoding}; fn object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } fn series(table_id: u32, tsid: u64) -> MetricSeriesId { diff --git a/src/mito2/src/sst/range_index/writer.rs b/src/mito2/src/sst/range_index/writer.rs index 4b7036803c..aa28cf1d30 100644 --- a/src/mito2/src/sst/range_index/writer.rs +++ b/src/mito2/src/sst/range_index/writer.rs @@ -537,7 +537,7 @@ mod tests { use crate::test_util::sst_util::{new_sparse_primary_key, sst_region_metadata_with_encoding}; fn object_store() -> ObjectStore { - ObjectStore::new(Memory::default()).unwrap().finish() + ObjectStore::new(Memory::default()).unwrap() } fn pk_schema(primary_key_type: DataType) -> SchemaRef { diff --git a/src/mito2/src/test_util.rs b/src/mito2/src/test_util.rs index 5bf55edb07..8b0302a365 100644 --- a/src/mito2/src/test_util.rs +++ b/src/mito2/src/test_util.rs @@ -568,7 +568,7 @@ impl TestEnv { .display() .to_string(); let builder = Fs::default(); - let object_store = ObjectStore::new(builder.root(&data_path)).unwrap().finish(); + let object_store = ObjectStore::new(builder.root(&data_path)).unwrap(); object_store_manager.add(storage_name, object_store); } let object_store_manager = Arc::new(object_store_manager); @@ -739,19 +739,16 @@ impl TestEnv { let object_store = if let Some(mock_layer) = self.object_store_mock_layer.as_ref() { debug!("create object store with mock layer"); - ObjectStore::new(builder) - .unwrap() - .layer(mock_layer.clone()) - .finish() + ObjectStore::new(builder).unwrap().layer(mock_layer.clone()) } else { - ObjectStore::new(builder).unwrap().finish() + ObjectStore::new(builder).unwrap() }; ObjectStoreManager::new("default", object_store) } pub(crate) fn create_in_memory_object_store_manager(&self) -> ObjectStoreManager { let builder = object_store::services::Memory::default(); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); ObjectStoreManager::new("memory", object_store) } @@ -771,11 +768,8 @@ impl TestEnv { ObjectStore::new(builder.root(&manifest_dir)) .unwrap() .layer(mock_layer.clone()) - .finish() } else { - ObjectStore::new(builder.root(&manifest_dir)) - .unwrap() - .finish() + ObjectStore::new(builder.root(&manifest_dir)).unwrap() }; // The "manifest_dir" here should be the relative path from the `object_store`'s root. diff --git a/src/mito2/src/test_util/scheduler_util.rs b/src/mito2/src/test_util/scheduler_util.rs index 10d4bffec5..9194094873 100644 --- a/src/mito2/src/test_util/scheduler_util.rs +++ b/src/mito2/src/test_util/scheduler_util.rs @@ -66,7 +66,7 @@ impl SchedulerEnv { let intm_mgr = IntermediateManager::init_fs(index_aux_path.to_str().unwrap()) .await .unwrap(); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); let access_layer = Arc::new(AccessLayer::new( "", PathType::Bare, diff --git a/src/object-store/Cargo.toml b/src/object-store/Cargo.toml index 8e6febf32e..f9b367c3dc 100644 --- a/src/object-store/Cargo.toml +++ b/src/object-store/Cargo.toml @@ -27,7 +27,7 @@ derive_builder = { workspace = true, optional = true } futures.workspace = true humantime-serde.workspace = true lazy_static.workspace = true -opendal = { version = "0.57", features = [ +opendal = { version = "0.58", features = [ "layers-tracing", "layers-prometheus", "services-azblob", @@ -37,6 +37,7 @@ opendal = { version = "0.57", features = [ "services-oss", "services-s3", ] } +opendal-http-transport-reqwest = "0.58" prometheus.workspace = true reqwest.workspace = true serde.workspace = true diff --git a/src/object-store/src/factory.rs b/src/object-store/src/factory.rs index 005817d269..9bdd656a07 100644 --- a/src/object-store/src/factory.rs +++ b/src/object-store/src/factory.rs @@ -15,7 +15,6 @@ use std::{fs, path}; use common_telemetry::info; -use opendal::layers::HttpClientLayer; #[cfg(feature = "mysql-object-store")] use opendal::services::Mysql; use opendal::services::{Fs, Gcs, Oss, S3}; @@ -26,7 +25,7 @@ use crate::config::MysqlConfig; use crate::config::{AzblobConfig, FileConfig, GcsConfig, ObjectStoreConfig, OssConfig, S3Config}; use crate::error::{self, Result}; use crate::services::Azblob; -use crate::util::{build_http_client, clean_temp_dir, join_dir, normalize_dir}; +use crate::util::{build_http_context, clean_temp_dir, join_dir, normalize_dir}; use crate::{ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR, ObjectStore, util}; pub async fn new_raw_object_store( @@ -55,9 +54,7 @@ pub async fn new_mysql_object_store(mysql_config: &MysqlConfig) -> Result Result .root(data_home) .atomic_write_dir(&atomic_write_dir); - let object_store = ObjectStore::new(builder) - .context(error::InitBackendSnafu)? - .finish(); + let object_store = ObjectStore::new(builder).context(error::InitBackendSnafu)?; Ok(object_store) } @@ -93,12 +88,11 @@ pub async fn new_azblob_object_store(azblob_config: &AzblobConfig) -> Result Result gcs_config.connection.bucket, &root ); - let client = build_http_client(&gcs_config.http_client)?; + let ctx = build_http_context(&gcs_config.http_client)?; let builder = Gcs::from(&gcs_config.connection); let operator = ObjectStore::new(builder) .context(error::InitBackendSnafu)? - .layer(HttpClientLayer::new(client)) - .finish(); + .with_context(ctx); Ok(operator) } @@ -127,12 +120,11 @@ pub async fn new_oss_object_store(oss_config: &OssConfig) -> Result oss_config.connection.bucket, &root ); - let client = build_http_client(&oss_config.http_client)?; + let ctx = build_http_context(&oss_config.http_client)?; let builder = Oss::from(&oss_config.connection); let operator = ObjectStore::new(builder) .context(error::InitBackendSnafu)? - .layer(HttpClientLayer::new(client)) - .finish(); + .with_context(ctx); Ok(operator) } @@ -144,12 +136,11 @@ pub async fn new_s3_object_store(s3_config: &S3Config) -> Result { s3_config.connection.bucket, &root ); - let client = build_http_client(&s3_config.http_client)?; + let ctx = build_http_context(&s3_config.http_client)?; let builder = S3::from(&s3_config.connection); let operator = ObjectStore::new(builder) .context(error::InitBackendSnafu)? - .layer(HttpClientLayer::new(client)) - .finish(); + .with_context(ctx); Ok(operator) } diff --git a/src/object-store/src/layers/mock.rs b/src/object-store/src/layers/mock.rs index f4d3df54d7..6b919aabb6 100644 --- a/src/object-store/src/layers/mock.rs +++ b/src/object-store/src/layers/mock.rs @@ -18,17 +18,15 @@ use std::sync::Arc; use derive_builder::Builder; pub use oio::*; pub use opendal::raw::{ - Access, Layer, LayeredAccess, OpDelete, OpList, OpRead, OpWrite, RpDelete, RpList, RpRead, - RpWrite, oio, + Layer, OpCopy, OpDelete, OpList, OpRead, OpWrite, RpRead, Service, Servicer, oio, }; -use opendal::raw::{OpCopier, OpCopy, RpCopy}; -pub use opendal::{Buffer, Error, ErrorKind, Metadata, Result}; +pub use opendal::{Buffer, BytesRange, Error, ErrorKind, Metadata, OperationContext, Result}; pub type MockWriterFactory = Arc oio::Writer + Send + Sync>; pub type MockReaderFactory = Arc oio::Reader + Send + Sync>; pub type MockListerFactory = Arc oio::Lister + Send + Sync>; pub type MockDeleterFactory = Arc oio::Deleter + Send + Sync>; -pub type CopyInterceptor = Arc Option> + Send + Sync>; +pub type CopyInterceptor = Arc Option> + Send + Sync>; #[derive(Builder)] pub struct MockLayer { @@ -56,23 +54,27 @@ impl Clone for MockLayer { } } -impl Layer for MockLayer { - type LayeredAccess = MockAccessor; +impl Debug for MockLayer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MockLayer").finish_non_exhaustive() + } +} - fn layer(&self, inner: A) -> Self::LayeredAccess { - MockAccessor { +impl Layer for MockLayer { + fn apply_service(&self, inner: Servicer) -> Servicer { + Arc::new(MockService { inner, writer_factory: self.writer_factory.clone(), reader_factory: self.reader_factory.clone(), lister_factory: self.lister_factory.clone(), deleter_factory: self.deleter_factory.clone(), copy_interceptor: self.copy_interceptor.clone(), - } + }) } } -pub struct MockAccessor { - inner: A, +struct MockService { + inner: Servicer, writer_factory: Option, reader_factory: Option, lister_factory: Option, @@ -80,11 +82,120 @@ pub struct MockAccessor { copy_interceptor: Option, } -impl Debug for MockAccessor { +impl Debug for MockService { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("MockAccessor") + f.debug_struct("MockService") .field("inner", &self.inner) - .finish() + .finish_non_exhaustive() + } +} + +impl Service for MockService { + type Reader = oio::Reader; + type Writer = oio::Writer; + type Lister = oio::Lister; + type Deleter = oio::Deleter; + type Copier = oio::Copier; + + fn info(&self) -> opendal::raw::ServiceInfo { + self.inner.info() + } + + fn capability(&self) -> opendal::Capability { + self.inner.capability() + } + + async fn create_dir( + &self, + ctx: &OperationContext, + path: &str, + args: opendal::raw::OpCreateDir, + ) -> Result { + self.inner.create_dir(ctx, path, args).await + } + + async fn stat( + &self, + ctx: &OperationContext, + path: &str, + args: opendal::raw::OpStat, + ) -> Result { + self.inner.stat(ctx, path, args).await + } + + fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result { + let reader = self.inner.read(ctx, path, args.clone())?; + if let Some(reader_factory) = self.reader_factory.as_ref() { + Ok(reader_factory(path, args, reader)) + } else { + Ok(reader) + } + } + + fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result { + let writer = self.inner.write(ctx, path, args.clone())?; + if let Some(writer_factory) = self.writer_factory.as_ref() { + Ok(writer_factory(path, args, writer)) + } else { + Ok(writer) + } + } + + fn delete(&self, ctx: &OperationContext) -> Result { + let deleter = self.inner.delete(ctx)?; + if let Some(deleter_factory) = self.deleter_factory.as_ref() { + Ok(deleter_factory(deleter)) + } else { + Ok(deleter) + } + } + + fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result { + let lister = self.inner.list(ctx, path, args.clone())?; + if let Some(lister_factory) = self.lister_factory.as_ref() { + Ok(lister_factory(path, args, lister)) + } else { + Ok(lister) + } + } + + fn copy( + &self, + ctx: &OperationContext, + from: &str, + to: &str, + args: OpCopy, + opts: opendal::raw::OpCopier, + ) -> Result { + if let Some(result) = self + .copy_interceptor + .as_ref() + .and_then(|copy_interceptor| copy_interceptor(from, to, args.clone())) + { + result?; + return Ok(Box::new(oio::OneShotCopier::completed()) as oio::Copier); + } + + self.inner.copy(ctx, from, to, args, opts) + } + + async fn rename( + &self, + ctx: &OperationContext, + from: &str, + to: &str, + args: opendal::raw::OpRename, + ) -> Result { + self.inner.rename(ctx, from, to, args).await + } + + async fn presign( + &self, + ctx: &OperationContext, + path: &str, + args: opendal::raw::OpPresign, + ) -> Result { + self.inner.presign(ctx, path, args).await } } @@ -93,8 +204,15 @@ pub struct MockReader { } impl oio::Read for MockReader { - async fn read(&mut self) -> Result { - self.inner.read().await + async fn open( + &self, + range: opendal::BytesRange, + ) -> Result<(opendal::raw::RpRead, Box)> { + self.inner.open(range).await + } + + async fn read(&self, range: opendal::BytesRange) -> Result<(opendal::raw::RpRead, Buffer)> { + self.inner.read(range).await } } @@ -139,108 +257,3 @@ impl oio::Delete for MockDeleter { self.inner.close().await } } - -impl LayeredAccess for MockAccessor { - type Inner = A; - type Reader = MockReader; - type Writer = MockWriter; - type Lister = MockLister; - type Deleter = MockDeleter; - type Copier = oio::Copier; - - fn inner(&self) -> &Self::Inner { - &self.inner - } - - async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> { - if let Some(reader_factory) = self.reader_factory.as_ref() { - let (rp_read, reader) = self.inner.read(path, args.clone()).await?; - let reader = reader_factory(path, args, Box::new(reader)); - Ok((rp_read, MockReader { inner: reader })) - } else { - self.inner.read(path, args).await.map(|(rp_read, reader)| { - ( - rp_read, - MockReader { - inner: Box::new(reader), - }, - ) - }) - } - } - - async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> { - if let Some(writer_factory) = self.writer_factory.as_ref() { - let (rp_write, writer) = self.inner.write(path, args.clone()).await?; - let writer = writer_factory(path, args, Box::new(writer)); - Ok((rp_write, MockWriter { inner: writer })) - } else { - self.inner - .write(path, args) - .await - .map(|(rp_write, writer)| { - ( - rp_write, - MockWriter { - inner: Box::new(writer), - }, - ) - }) - } - } - - async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> { - if let Some(deleter_factory) = self.deleter_factory.as_ref() { - let (rp_delete, deleter) = self.inner.delete().await?; - let deleter = deleter_factory(Box::new(deleter)); - Ok((rp_delete, MockDeleter { inner: deleter })) - } else { - self.inner.delete().await.map(|(rp_delete, deleter)| { - ( - rp_delete, - MockDeleter { - inner: Box::new(deleter), - }, - ) - }) - } - } - - async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> { - if let Some(lister_factory) = self.lister_factory.as_ref() { - let (rp_list, lister) = self.inner.list(path, args.clone()).await?; - let lister = lister_factory(path, args, Box::new(lister)); - Ok((rp_list, MockLister { inner: lister })) - } else { - self.inner.list(path, args).await.map(|(rp_list, lister)| { - ( - rp_list, - MockLister { - inner: Box::new(lister), - }, - ) - }) - } - } - - async fn copy( - &self, - from: &str, - to: &str, - args: OpCopy, - opts: OpCopier, - ) -> Result<(RpCopy, Self::Copier)> { - if let Some(result) = self - .copy_interceptor - .as_ref() - .and_then(|copy_interceptor| copy_interceptor(from, to, args.clone())) - { - return result.map(|rp_copy| (rp_copy, Box::new(()) as oio::Copier)); - } - - self.inner - .copy(from, to, args, opts) - .await - .map(|(rp_copy, copier)| (rp_copy, Box::new(copier) as oio::Copier)) - } -} diff --git a/src/object-store/src/lib.rs b/src/object-store/src/lib.rs index a9257c2423..05c344485c 100644 --- a/src/object-store/src/lib.rs +++ b/src/object-store/src/lib.rs @@ -12,10 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub use opendal::raw::{Access, HttpClient}; +pub use opendal::raw::{Service, ServiceInfo, Servicer}; pub use opendal::{ Buffer, Builder as ObjectStoreBuilder, Entry, EntryMode, Error, ErrorKind, FuturesAsyncReader, - FuturesAsyncWriter, Lister, Operator as ObjectStore, Reader, Result, Writer, services, + FuturesAsyncWriter, HttpTransporter, Lister, OperationContext, Operator as ObjectStore, Reader, + Result, Writer, services, }; pub mod config; diff --git a/src/object-store/src/manager.rs b/src/object-store/src/manager.rs index 3753640824..455eda3b6f 100644 --- a/src/object-store/src/manager.rs +++ b/src/object-store/src/manager.rs @@ -68,7 +68,7 @@ mod tests { fn new_object_store(dir: &TempDir) -> ObjectStore { let store_dir = dir.path().to_str().unwrap(); let builder = Builder::default().root(store_dir); - ObjectStore::new(builder).unwrap().finish() + ObjectStore::new(builder).unwrap() } #[test] diff --git a/src/object-store/src/secure_fs.rs b/src/object-store/src/secure_fs.rs index d793dee6e3..581cc2837c 100644 --- a/src/object-store/src/secure_fs.rs +++ b/src/object-store/src/secure_fs.rs @@ -21,9 +21,11 @@ use std::{fmt, io}; use cap_std::ambient_authority; use cap_std::fs::{Dir, DirEntry, OpenOptions, ReadDir}; +use opendal::layers::SimulateLayer; use opendal::raw::*; use opendal::{ - Buffer, Capability, EntryMode, Error, ErrorKind, Metadata, Operator, OperatorBuilder, Result, + Buffer, BytesRange, Capability, EntryMode, Error, ErrorKind, Metadata, OperationContext, + Operator, Result, }; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; @@ -102,7 +104,11 @@ impl SecureFsRoot { /// Builds an OpenDAL operator confined to this root. pub fn build_operator(&self) -> Operator { - OperatorBuilder::new(SecureFsBackend::new(self.clone())).finish() + Operator::from_parts( + OperationContext::default(), + Arc::new(SecureFsBackend::new(self.clone())) as Servicer, + ) + .layer(SimulateLayer::default()) } } @@ -163,48 +169,57 @@ fn metadata_from_fs(metadata: cap_std::fs::Metadata) -> Result { #[derive(Clone, Debug)] struct SecureFsBackend { root: SecureFsRoot, - info: Arc, + info: ServiceInfo, + capability: Capability, } impl SecureFsBackend { fn new(root: SecureFsRoot) -> Self { - let info = AccessorInfo::default(); - info.set_scheme("fs") - .set_root(&root.path().to_string_lossy()) - .set_native_capability(Capability { - stat: true, - read: true, - write: true, - write_can_empty: true, - write_can_append: true, - write_can_multi: true, - write_with_if_not_exists: true, - create_dir: true, - delete: true, - delete_with_recursive: true, - list: true, - shared: true, - ..Default::default() - }); + let info = ServiceInfo::new("fs", root.path().to_string_lossy(), ""); + let capability = Capability { + stat: true, + read: true, + write: true, + write_can_empty: true, + write_can_append: true, + write_can_multi: true, + write_with_if_not_exists: true, + create_dir: true, + delete: true, + delete_with_recursive: true, + list: true, + shared: true, + ..Default::default() + }; Self { root, - info: info.into(), + info, + capability, } } } -impl Access for SecureFsBackend { - type Reader = SecureFsReader; +impl Service for SecureFsBackend { + type Reader = oio::StreamReader; type Writer = SecureFsWriter; - type Lister = Option; + type Lister = SecureFsLister; type Deleter = oio::OneShotDeleter; type Copier = (); - fn info(&self) -> Arc { + fn info(&self) -> ServiceInfo { self.info.clone() } - async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result { + fn capability(&self) -> Capability { + self.capability + } + + async fn create_dir( + &self, + _: &OperationContext, + path: &str, + _: OpCreateDir, + ) -> Result { let path = backend_path(path).map_err(new_std_io_error)?; let root = self.root.clone(); common_runtime::spawn_blocking_global(move || root.dir.create_dir_all(path)) @@ -214,7 +229,7 @@ impl Access for SecureFsBackend { Ok(RpCreateDir::default()) } - async fn stat(&self, path: &str, _: OpStat) -> Result { + async fn stat(&self, _: &OperationContext, path: &str, _: OpStat) -> Result { let path = backend_path(path).map_err(new_std_io_error)?; let root = self.root.clone(); let metadata = common_runtime::spawn_blocking_global(move || { @@ -230,139 +245,118 @@ impl Access for SecureFsBackend { Ok(RpStat::new(metadata_from_fs(metadata)?)) } - async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> { + fn read(&self, _: &OperationContext, path: &str, _: OpRead) -> Result { let path = backend_path(path).map_err(new_std_io_error)?; - let root = self.root.clone(); - let file = common_runtime::spawn_blocking_global(move || root.dir.open(path)) - .await - .map_err(new_task_join_error)? - .map_err(new_std_io_error)?; - let mut file = tokio::fs::File::from_std(file.into_std()); - if args.range().offset() != 0 { - file.seek(io::SeekFrom::Start(args.range().offset())) - .await - .map_err(new_std_io_error)?; - } - Ok(( - RpRead::default(), - SecureFsReader { - file, - remaining: args.range().size().unwrap_or(u64::MAX), - }, - )) + Ok(oio::StreamReader::new(SecureFsReader { + root: self.root.clone(), + path, + })) } - async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> { + fn write(&self, _: &OperationContext, path: &str, args: OpWrite) -> Result { let path = backend_path(path).map_err(new_std_io_error)?; - let root = self.root.clone(); - let if_not_exists = args.if_not_exists(); - let file = common_runtime::spawn_blocking_global(move || { - if let Some(parent) = path.parent() - && !parent.as_os_str().is_empty() - { - root.dir.create_dir_all(parent).map_err(new_std_io_error)?; - } - - let mut options = OpenOptions::new(); - options.write(true); - if args.if_not_exists() { - options.create_new(true); - } else { - options.create(true); - } - if args.append() { - options.append(true); - } else { - options.truncate(true); - } - root.dir - .open_with(path, &options) - .map_err(|error| parse_write_error(error, if_not_exists)) + Ok(SecureFsWriter { + root: self.root.clone(), + path, + args, + file: None, }) - .await - .map_err(new_task_join_error)??; - - Ok(( - RpWrite::default(), - SecureFsWriter { - file: tokio::fs::File::from_std(file.into_std()), - }, - )) } - async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> { - Ok(( - RpDelete::default(), - oio::OneShotDeleter::new(SecureFsDeleter { - root: self.root.clone(), - }), - )) + fn delete(&self, _: &OperationContext) -> Result { + Ok(oio::OneShotDeleter::new(SecureFsDeleter { + root: self.root.clone(), + })) } - async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> { + fn list(&self, _: &OperationContext, path: &str, _: OpList) -> Result { let path = backend_path(path).map_err(new_std_io_error)?; let display_prefix = if path.as_os_str().is_empty() { String::new() } else { format!("{}/", path.to_string_lossy().replace('\\', "/")) }; - let root = self.root.clone(); - let read_dir = common_runtime::spawn_blocking_global(move || { - let result = (|| { - let dir = if path.as_os_str().is_empty() { - root.dir.open_dir(".")? - } else { - root.dir.open_dir(&path)? - }; - dir.entries() - })(); - - match result { - Ok(read_dir) => Ok(Some(read_dir)), - Err(error) - if matches!( - error.kind(), - io::ErrorKind::NotFound | io::ErrorKind::NotADirectory - ) => - { - Ok(None) - } - Err(error) => Err(error), - } + Ok(SecureFsLister { + root: self.root.clone(), + path, + display_prefix, + read_dir: None, + entries: vec![].into_iter(), + seeded: false, + done: false, }) - .await - .map_err(new_task_join_error)? - .map_err(new_std_io_error)?; + } - let Some(read_dir) = read_dir else { - return Ok((RpList::default(), None)); - }; - let current_path = oio::Entry::new( - if display_prefix.is_empty() { - "/" - } else { - &display_prefix - }, - Metadata::new(EntryMode::DIR), - ); - Ok(( - RpList::default(), - Some(SecureFsLister { - read_dir: Arc::new(Mutex::new(read_dir)), - display_prefix, - entries: vec![current_path].into_iter(), - done: false, - }), + fn copy( + &self, + _: &OperationContext, + _: &str, + _: &str, + _: OpCopy, + _: OpCopier, + ) -> Result { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + async fn rename( + &self, + _: &OperationContext, + _: &str, + _: &str, + _: OpRename, + ) -> Result { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", + )) + } + + async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result { + Err(Error::new( + ErrorKind::Unsupported, + "operation is not supported", )) } } struct SecureFsReader { + root: SecureFsRoot, + path: PathBuf, +} + +impl oio::StreamRead for SecureFsReader { + async fn open(&self, range: BytesRange) -> Result<(RpRead, Box)> { + let path = self.path.clone(); + let root = self.root.clone(); + let file = common_runtime::spawn_blocking_global(move || root.dir.open(path)) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error)?; + let mut file = tokio::fs::File::from_std(file.into_std()); + if range.offset() != 0 { + file.seek(io::SeekFrom::Start(range.offset())) + .await + .map_err(new_std_io_error)?; + } + Ok(( + RpRead::default(), + Box::new(SecureFsReadStream { + file, + remaining: range.size().unwrap_or(u64::MAX), + }), + )) + } +} + +struct SecureFsReadStream { file: tokio::fs::File, remaining: u64, } -impl oio::Read for SecureFsReader { +impl oio::ReadStream for SecureFsReadStream { async fn read(&mut self) -> Result { if self.remaining == 0 { return Ok(Buffer::new()); @@ -382,21 +376,65 @@ impl oio::Read for SecureFsReader { } struct SecureFsWriter { - file: tokio::fs::File, + root: SecureFsRoot, + path: PathBuf, + args: OpWrite, + file: Option, +} + +impl SecureFsWriter { + async fn ensure_file(&mut self) -> Result<&mut tokio::fs::File> { + if self.file.is_none() { + let path = self.path.clone(); + let root = self.root.clone(); + let if_not_exists = self.args.if_not_exists(); + let append = self.args.append(); + let file = common_runtime::spawn_blocking_global(move || { + if let Some(parent) = path.parent() + && !parent.as_os_str().is_empty() + { + root.dir.create_dir_all(parent).map_err(new_std_io_error)?; + } + + let mut options = OpenOptions::new(); + options.write(true); + if if_not_exists { + options.create_new(true); + } else { + options.create(true); + } + if append { + options.append(true); + } else { + options.truncate(true); + } + root.dir + .open_with(path, &options) + .map_err(|error| parse_write_error(error, if_not_exists)) + }) + .await + .map_err(new_task_join_error)??; + + self.file = Some(tokio::fs::File::from_std(file.into_std())); + } + Ok(self.file.as_mut().expect("file must be initialized")) + } } impl oio::Write for SecureFsWriter { async fn write(&mut self, buffer: Buffer) -> Result<()> { - self.file + self.ensure_file() + .await? .write_all(&buffer.to_bytes()) .await .map_err(new_std_io_error) } async fn close(&mut self) -> Result { - self.file.flush().await.map_err(new_std_io_error)?; - self.file.sync_all().await.map_err(new_std_io_error)?; - let metadata = self.file.metadata().await.map_err(new_std_io_error)?; + let file = self.ensure_file().await?; + file.flush().await.map_err(new_std_io_error)?; + file.sync_all().await.map_err(new_std_io_error)?; + let metadata = file.metadata().await.map_err(new_std_io_error)?; Ok(Metadata::new(EntryMode::FILE) .with_content_length(metadata.len()) .with_last_modified(Timestamp::try_from( @@ -413,14 +451,65 @@ impl oio::Write for SecureFsWriter { } struct SecureFsLister { - read_dir: Arc>, + root: SecureFsRoot, + path: PathBuf, display_prefix: String, + read_dir: Option>>, entries: IntoIter, + seeded: bool, done: bool, } impl oio::List for SecureFsLister { async fn next(&mut self) -> Result> { + if !self.seeded { + self.seeded = true; + let path = self.path.clone(); + let root = self.root.clone(); + let display_prefix = self.display_prefix.clone(); + let read_dir = common_runtime::spawn_blocking_global(move || { + let result = (|| { + let dir = if path.as_os_str().is_empty() { + root.dir.open_dir(".")? + } else { + root.dir.open_dir(&path)? + }; + dir.entries() + })(); + + match result { + Ok(read_dir) => Ok(Some(read_dir)), + Err(error) + if matches!( + error.kind(), + io::ErrorKind::NotFound | io::ErrorKind::NotADirectory + ) => + { + Ok(None) + } + Err(error) => Err(error), + } + }) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error)?; + + let Some(read_dir) = read_dir else { + self.done = true; + return Ok(None); + }; + self.read_dir = Some(Arc::new(Mutex::new(read_dir))); + let current_path = oio::Entry::new( + if display_prefix.is_empty() { + "/" + } else { + &display_prefix + }, + Metadata::new(EntryMode::DIR), + ); + self.entries = vec![current_path].into_iter(); + } + if let Some(entry) = self.entries.next() { return Ok(Some(entry)); } @@ -428,7 +517,10 @@ impl oio::List for SecureFsLister { return Ok(None); } - let read_dir = self.read_dir.clone(); + let Some(read_dir) = self.read_dir.clone() else { + self.done = true; + return Ok(None); + }; let display_prefix = self.display_prefix.clone(); let (entries, done) = common_runtime::spawn_blocking_global(move || { let mut read_dir = read_dir @@ -539,12 +631,40 @@ impl oio::OneShotDelete for SecureFsDeleter { mod tests { use bytes::Bytes; use common_test_util::temp_dir::create_temp_dir; - use opendal::ErrorKind; use opendal::raw::oio::List; - use opendal::raw::{Access, OpList}; + use opendal::raw::{OpList, Service}; + use opendal::{BytesRange, ErrorKind, OperationContext}; use super::{LIST_BATCH_SIZE, SecureFsBackend, SecureFsRoot, read_list_entry}; + #[tokio::test] + async fn test_operator_suffix_reads_final_bytes() { + let temp_dir = create_temp_dir("secure_fs_operator_suffix"); + std::fs::write(temp_dir.path().join("file"), b"0123456789").unwrap(); + let operator = SecureFsRoot::open(temp_dir.path()) + .unwrap() + .build_operator(); + + assert_eq!( + Bytes::from_static(b"789"), + operator + .read_with("file") + .range(7..10) + .await + .unwrap() + .to_bytes() + ); + assert_eq!( + Bytes::from_static(b"789"), + operator + .read_with("file") + .range(BytesRange::Suffix { size: 3 }) + .await + .unwrap() + .to_bytes() + ); + } + #[tokio::test] async fn test_lister_streams_entries() { let temp_dir = create_temp_dir("secure_fs_lister_streams_entries"); @@ -554,10 +674,8 @@ mod tests { let root = SecureFsRoot::open(temp_dir.path()).unwrap(); let backend = SecureFsBackend::new(root); - let (_, lister) = backend.list("/", OpList::new()).await.unwrap(); - let mut lister = lister.unwrap(); - - assert_eq!(1, lister.entries.len()); + let ctx = OperationContext::default(); + let mut lister = backend.list(&ctx, "/", OpList::new()).unwrap(); let mut paths = Vec::new(); while let Some(entry) = lister.next().await.unwrap() { diff --git a/src/object-store/src/util.rs b/src/object-store/src/util.rs index 92f0bd7299..019f089c5e 100644 --- a/src/object-store/src/util.rs +++ b/src/object-store/src/util.rs @@ -17,12 +17,13 @@ use std::path; use common_error::root_source; use common_telemetry::{debug, error, info, warn}; -use opendal::ErrorKind; use opendal::layers::{ LoggingInterceptor, LoggingLayer, RetryEvent, RetryInterceptor, RetryLayer, TracingLayer, }; -use opendal::raw::{AccessorInfo, HttpClient, Operation}; +use opendal::raw::{Operation, ServiceInfo}; use opendal::services::FS_SCHEME; +use opendal::{ErrorKind, HttpTransporter, OperationContext}; +use opendal_http_transport_reqwest::ReqwestTransport; use snafu::ResultExt; use crate::config::HttpClientConfig; @@ -173,7 +174,7 @@ impl LoggingInterceptor for DefaultLoggingInterceptor { #[inline] fn log( &self, - info: &AccessorInfo, + info: &ServiceInfo, operation: Operation, context: &[(&str, &str)], message: &str, @@ -211,7 +212,8 @@ impl LoggingInterceptor for DefaultLoggingInterceptor { } } -pub(crate) fn build_http_client(config: &HttpClientConfig) -> error::Result { +/// Builds an [`OperationContext`] with a custom HTTP transport from `config`. +pub(crate) fn build_http_context(config: &HttpClientConfig) -> error::Result { if config.skip_ssl_validation { common_telemetry::warn!( "Skipping SSL validation for object storage HTTP client. Please ensure the environment is trusted." @@ -226,7 +228,8 @@ pub(crate) fn build_http_client(config: &HttpClientConfig) -> error::Result error::Result<()> { @@ -302,9 +305,7 @@ mod tests { #[test] fn test_fs_is_not_object_storage() { - let object_store = ObjectStore::new(Fs::default().root("/tmp")) - .unwrap() - .finish(); + let object_store = ObjectStore::new(Fs::default().root("/tmp")).unwrap(); assert_eq!(FS_SCHEME, object_store.info().scheme()); assert!(!is_object_storage(&object_store)); diff --git a/src/object-store/tests/object_store_test.rs b/src/object-store/tests/object_store_test.rs index d373972473..9cf03fe42c 100644 --- a/src/object-store/tests/object_store_test.rs +++ b/src/object-store/tests/object_store_test.rs @@ -114,7 +114,7 @@ async fn test_object_list_start_after(store: &ObjectStore) -> Result<()> { // `start_after` is a service-level capability. Skip the checks when the // backend (e.g. the local Fs service) doesn't honor it natively — the // bound would be silently ignored and the full listing returned. - if !store.info().native_capability().list_with_start_after { + if !store.info().capability().list_with_start_after { info!("Skip test_object_list_start_after: backend {scheme} lacks start_after support"); return Ok(()); } @@ -231,7 +231,7 @@ fn create_temp_dir(prefix: &str) -> Result { #[tokio::test] async fn test_opendal_memory_smoke() -> Result<()> { - let op = opendal::Operator::new(Memory::default())?.finish(); + let op = opendal::Operator::new(Memory::default())?; let store: OpendalStore = OpendalStore::new(op); assert_eq!("memory", store.info().scheme()); assert!(format!("{store}").contains("memory")); @@ -262,7 +262,7 @@ async fn test_fs_backend() -> Result<()> { .root(&data_dir.path().to_string_lossy()) .atomic_write_dir(&tmp_dir.path().to_string_lossy()); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let store = object_store::util::with_instrument_layers(store, false); test_object_crud(&store).await?; @@ -291,7 +291,7 @@ async fn test_s3_backend() -> Result<()> { .region(&env::var("GT_S3_REGION")?) .bucket(&bucket); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let store = object_store::util::with_instrument_layers(store, false); let guard = TempFolder::new(&store, "/"); @@ -321,7 +321,7 @@ async fn test_oss_backend() -> Result<()> { .access_key_secret(&env::var("GT_OSS_ACCESS_KEY")?) .bucket(&bucket); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let store = object_store::util::with_instrument_layers(store, false); let guard = TempFolder::new(&store, "/"); @@ -351,7 +351,7 @@ async fn test_azblob_backend() -> Result<()> { .account_key(&env::var("GT_AZBLOB_ACCOUNT_KEY")?) .container(&container); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let store = object_store::util::with_instrument_layers(store, false); let guard = TempFolder::new(&store, "/"); @@ -380,7 +380,7 @@ async fn test_gcs_backend() -> Result<()> { .credential(&env::var("GT_GCS_CREDENTIAL").unwrap()) .endpoint(&env::var("GT_GCS_ENDPOINT").unwrap()); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); let store = object_store::util::with_instrument_layers(store, false); let guard = TempFolder::new(&store, "/"); diff --git a/src/operator/src/statement/copy_database.rs b/src/operator/src/statement/copy_database.rs index 3db6eb8e76..43b2692cdd 100644 --- a/src/operator/src/statement/copy_database.rs +++ b/src/operator/src/statement/copy_database.rs @@ -319,7 +319,7 @@ mod tests { let dir = common_test_util::temp_dir::create_temp_dir("test_list_files_to_copy"); let store_dir = normalize_dir(dir.path().to_str().unwrap()); let builder = Fs::default().root(&store_dir); - let object_store = ObjectStore::new(builder).unwrap().finish(); + let object_store = ObjectStore::new(builder).unwrap(); object_store.write("a.parquet", "").await.unwrap(); object_store.write("b.parquet", "").await.unwrap(); object_store.write("c.csv", "").await.unwrap(); diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs index cb958a84e5..9d12848ce6 100644 --- a/tests-integration/src/test_util.rs +++ b/tests-integration/src/test_util.rs @@ -182,7 +182,7 @@ pub fn get_test_store_config(store_type: &StorageType) -> (ObjectStoreConfig, Te let builder = Gcs::from(&gcs_config.connection); let config = ObjectStoreConfig::Gcs(gcs_config); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); (config, TempDirGuard::Gcs(TempFolder::new(&store, "/"))) } StorageType::Azblob => { @@ -200,7 +200,7 @@ pub fn get_test_store_config(store_type: &StorageType) -> (ObjectStoreConfig, Te let builder = Azblob::from(&azblob_config.connection); let config = ObjectStoreConfig::Azblob(azblob_config); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); (config, TempDirGuard::Azblob(TempFolder::new(&store, "/"))) } StorageType::Oss => { @@ -217,7 +217,7 @@ pub fn get_test_store_config(store_type: &StorageType) -> (ObjectStoreConfig, Te let builder = Oss::from(&oss_config.connection); let config = ObjectStoreConfig::Oss(oss_config); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); (config, TempDirGuard::Oss(TempFolder::new(&store, "/"))) } StorageType::S3 | StorageType::S3WithCache => { @@ -231,7 +231,7 @@ pub fn get_test_store_config(store_type: &StorageType) -> (ObjectStoreConfig, Te let builder = S3::from(&s3_config.connection); let config = ObjectStoreConfig::S3(s3_config); - let store = ObjectStore::new(builder).unwrap().finish(); + let store = ObjectStore::new(builder).unwrap(); (config, TempDirGuard::S3(TempFolder::new(&store, "/"))) } StorageType::File => (ObjectStoreConfig::File(FileConfig {}), TempDirGuard::None),