diff --git a/Cargo.lock b/Cargo.lock index 976f658a51..483b30cc0c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8465,6 +8465,7 @@ dependencies = [ "fs-err", "futures", "prost 0.14.4", + "prost-types 0.14.4", "tokio", "tonic", "tonic-prost", diff --git a/Cargo.toml b/Cargo.toml index 020556ff69..18a104ba5e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -259,6 +259,7 @@ pprof = { version = "0.15.0", features = ["flamegraph", "prost-codec"] } proptest = { version = "1.11.0", default-features = false, features = ["std"] } prost = "0.14.0" prost-build = { version = "0.14.0", features = ["cleanup-markdown"] } +prost-types = "0.14.0" prost-wkt-types = "0.7" prost-for-raft = { package = "prost", version = "=0.11.9" } # version of prost used by raft raft = { git = "https://github.com/tikv/raft-rs", rev = "aafb07c7bab439c6139926a77dfafc5b10e9bc84" ,features = ["prost-codec"], default-features = false } diff --git a/lib/api/src/grpc/conversions.rs b/lib/api/src/grpc/conversions.rs index f4072940f2..e6a7c8b49c 100644 --- a/lib/api/src/grpc/conversions.rs +++ b/lib/api/src/grpc/conversions.rs @@ -2780,6 +2780,10 @@ pub fn date_time_to_proto(date_time: DateTimePayloadType) -> prost_wkt_types::Ti naive_date_time_to_proto(date_time.0.naive_utc()) } +pub fn system_time_to_proto(time: std::time::SystemTime) -> prost_wkt_types::Timestamp { + naive_date_time_to_proto(chrono::DateTime::::from(time).naive_utc()) +} + pub fn try_date_time_from_proto( date_time: prost_wkt_types::Timestamp, ) -> Result { diff --git a/lib/api/src/grpc/proto/storage_read_service.proto b/lib/api/src/grpc/proto/storage_read_service.proto index db8e98f2d0..6c6f4d426e 100644 --- a/lib/api/src/grpc/proto/storage_read_service.proto +++ b/lib/api/src/grpc/proto/storage_read_service.proto @@ -2,6 +2,8 @@ syntax = "proto3"; package qdrant; +import "google/protobuf/timestamp.proto"; + option csharp_namespace = "Qdrant.Client.Grpc"; // Shard-scoped raw I/O over on-disk collection storage. @@ -53,6 +55,8 @@ message FileExistsResponse { message ListFilesEntry { string path = 1; uint64 size = 2; + // Last modification time, when the underlying storage exposes one. + optional google.protobuf.Timestamp last_modified = 3; } message ListFilesResponse { diff --git a/lib/api/src/grpc/qdrant.rs b/lib/api/src/grpc/qdrant.rs index 86b15c9a46..119abcd84a 100644 --- a/lib/api/src/grpc/qdrant.rs +++ b/lib/api/src/grpc/qdrant.rs @@ -16273,6 +16273,9 @@ pub struct ListFilesEntry { pub path: ::prost::alloc::string::String, #[prost(uint64, tag = "2")] pub size: u64, + /// Last modification time, when the underlying storage exposes one. + #[prost(message, optional, tag = "3")] + pub last_modified: ::core::option::Option<::prost_wkt_types::Timestamp>, } #[derive(serde::Serialize)] #[derive(Clone, PartialEq, ::prost::Message)] diff --git a/lib/common/common/src/universal_io/cached_fs/mod.rs b/lib/common/common/src/universal_io/cached_fs/mod.rs index de1abc45cb..3315466579 100644 --- a/lib/common/common/src/universal_io/cached_fs/mod.rs +++ b/lib/common/common/src/universal_io/cached_fs/mod.rs @@ -16,6 +16,8 @@ use crate::universal_io::{ pub struct FileInfo { /// Length in bytes of the entire file pub size: u64, + /// Last modification time, when the listing backend exposes one + pub last_modified: Option, } /// Read-only filesystem wrapper that snapshots the file listing and serves @@ -128,6 +130,7 @@ impl CachedFs { .map(|(path, info)| ListedFile { path: path.clone(), size: info.size, + last_modified: info.last_modified, }) .collect() } @@ -140,10 +143,19 @@ impl CachedReadFs for CachedFs { let files_info: HashMap<_, _> = list .into_iter() - .map(|ListedFile { path, size }| { - let info = FileInfo { size }; - (path, info) - }) + .map( + |ListedFile { + path, + size, + last_modified, + }| { + let info = FileInfo { + size, + last_modified, + }; + (path, info) + }, + ) .collect(); self.files_info = Some(files_info); diff --git a/lib/common/common/src/universal_io/local_file_ops.rs b/lib/common/common/src/universal_io/local_file_ops.rs index d7c5e0b4fc..c081c7b44b 100644 --- a/lib/common/common/src/universal_io/local_file_ops.rs +++ b/lib/common/common/src/universal_io/local_file_ops.rs @@ -65,10 +65,11 @@ pub fn local_list_files(prefix_path: &Path) -> crate::universal_io::Result crate::universal_io::Result { pub struct ListedFile { pub path: std::path::PathBuf, pub size: u64, + /// Last modification time, when the backend exposes one (local + /// filesystems, object stores); `None` otherwise. + pub last_modified: Option, } pub type ByteOffset = u64; diff --git a/lib/common/io_bridge_object_store/src/source.rs b/lib/common/io_bridge_object_store/src/source.rs index 11b1fd2467..5f13d6aead 100644 --- a/lib/common/io_bridge_object_store/src/source.rs +++ b/lib/common/io_bridge_object_store/src/source.rs @@ -17,6 +17,7 @@ use std::future::Future; use std::ops::Range; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::SystemTime; use bytes::Bytes; use common::universal_io::{ListedFile, Result, UniversalIoError, UniversalKind}; @@ -89,6 +90,7 @@ impl AsyncRead for ObjectStoreSource { location.starts_with(&prefix_str).then(|| ListedFile { path: PathBuf::from(location), size: e.size, + last_modified: Some(SystemTime::from(e.last_modified)), }) }) .collect()), @@ -320,7 +322,13 @@ mod tests { .block_on(source.list_files(Path::new("dir/page_"))) .expect("list_files") .into_iter() - .map(|ListedFile { path, size }| (path.to_string_lossy().into_owned(), size)) + .map( + |ListedFile { + path, + size, + last_modified: _, + }| (path.to_string_lossy().into_owned(), size), + ) .collect(); files.sort(); assert_eq!( diff --git a/lib/common/io_bridge_object_store/src/tests/integration.rs b/lib/common/io_bridge_object_store/src/tests/integration.rs index aae1b1de14..817eab3701 100644 --- a/lib/common/io_bridge_object_store/src/tests/integration.rs +++ b/lib/common/io_bridge_object_store/src/tests/integration.rs @@ -130,7 +130,12 @@ fn test_list_files() { .block_on(store.list_files(Path::new("listed"))) .expect("list_files"); assert_eq!(files.len(), 3); - for ListedFile { path, size } in &files { + for ListedFile { + path, + size, + last_modified: _, + } in &files + { assert!(path.to_string_lossy().starts_with("listed/")); assert_eq!(*size, 1); } diff --git a/lib/uio-grpc-client/Cargo.toml b/lib/uio-grpc-client/Cargo.toml index cbb1d568b6..c26ada3cdc 100644 --- a/lib/uio-grpc-client/Cargo.toml +++ b/lib/uio-grpc-client/Cargo.toml @@ -15,6 +15,7 @@ common = { path = "../common/common" } bytes = { workspace = true } futures = { workspace = true } prost = { workspace = true } +prost-types = { workspace = true } tokio = { workspace = true } tonic = { workspace = true } tonic-prost = { workspace = true } diff --git a/lib/uio-grpc-client/proto/storage_read_service.proto b/lib/uio-grpc-client/proto/storage_read_service.proto index db8e98f2d0..6c6f4d426e 100644 --- a/lib/uio-grpc-client/proto/storage_read_service.proto +++ b/lib/uio-grpc-client/proto/storage_read_service.proto @@ -2,6 +2,8 @@ syntax = "proto3"; package qdrant; +import "google/protobuf/timestamp.proto"; + option csharp_namespace = "Qdrant.Client.Grpc"; // Shard-scoped raw I/O over on-disk collection storage. @@ -53,6 +55,8 @@ message FileExistsResponse { message ListFilesEntry { string path = 1; uint64 size = 2; + // Last modification time, when the underlying storage exposes one. + optional google.protobuf.Timestamp last_modified = 3; } message ListFilesResponse { diff --git a/lib/uio-grpc-client/src/generated/qdrant.rs b/lib/uio-grpc-client/src/generated/qdrant.rs index df6cea7e01..5cb7b57350 100644 --- a/lib/uio-grpc-client/src/generated/qdrant.rs +++ b/lib/uio-grpc-client/src/generated/qdrant.rs @@ -29,6 +29,9 @@ pub struct ListFilesEntry { pub path: ::prost::alloc::string::String, #[prost(uint64, tag = "2")] pub size: u64, + /// Last modification time, when the underlying storage exposes one. + #[prost(message, optional, tag = "3")] + pub last_modified: ::core::option::Option<::prost_types::Timestamp>, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct ListFilesResponse { diff --git a/lib/uio-grpc-client/src/read.rs b/lib/uio-grpc-client/src/read.rs index 7cc57a96cc..e521b83c97 100644 --- a/lib/uio-grpc-client/src/read.rs +++ b/lib/uio-grpc-client/src/read.rs @@ -147,6 +147,9 @@ impl Client { .map(|entry| ListedFile { path: PathBuf::from(entry.path), size: entry.size, + last_modified: entry + .last_modified + .and_then(|ts| std::time::SystemTime::try_from(ts).ok()), }) .collect()) } diff --git a/lib/uio-grpc-client/src/tests.rs b/lib/uio-grpc-client/src/tests.rs index d819ce5b11..d346d84f83 100644 --- a/lib/uio-grpc-client/src/tests.rs +++ b/lib/uio-grpc-client/src/tests.rs @@ -13,6 +13,11 @@ use crate::generated::qdrant::storage_read_server::{ use crate::generated::qdrant::*; use crate::read::Client; +/// Fixed modification time served by the mock for every listed file. +fn mock_modified_time() -> std::time::SystemTime { + std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000) +} + struct MockServer { files: Arc>>, } @@ -44,6 +49,10 @@ impl StorageReadTrait for MockServer { ListFilesEntry { path: path.clone(), size: data.len() as u64, + last_modified: Some(prost_types::Timestamp { + seconds: 1_000_000, + nanos: 0, + }), } }) }) @@ -206,10 +215,12 @@ async fn list_files() { ListedFile { path: PathBuf::from("data/a.bin"), size: 10, + last_modified: Some(mock_modified_time()), }, ListedFile { path: PathBuf::from("data/b.bin"), size: 10, + last_modified: Some(mock_modified_time()), }, ] ); diff --git a/src/tonic/api/storage_read_api/mod.rs b/src/tonic/api/storage_read_api/mod.rs index 177808dd87..28d8592ff7 100644 --- a/src/tonic/api/storage_read_api/mod.rs +++ b/src/tonic/api/storage_read_api/mod.rs @@ -1,6 +1,7 @@ use std::pin::Pin; use std::sync::Arc; +use api::grpc::conversions::system_time_to_proto; use api::grpc::qdrant::storage_read_server::StorageRead; use api::grpc::qdrant::{ FileExistsRequest, FileExistsResponse, FileLengthRequest, FileLengthResponse, ListFilesEntry, @@ -95,19 +96,26 @@ where let files = files .into_iter() - .filter_map(|ListedFile { path, size }| { - path.strip_prefix(&base).ok().map(|rel| { - // Always use forward slashes in gRPC responses regardless of OS. - let components = rel - .components() - .filter_map(|c| c.as_os_str().to_str()) - .collect::>(); - ListFilesEntry { - path: components.join("/"), - size, - } - }) - }) + .filter_map( + |ListedFile { + path, + size, + last_modified, + }| { + path.strip_prefix(&base).ok().map(|rel| { + // Always use forward slashes in gRPC responses regardless of OS. + let components = rel + .components() + .filter_map(|c| c.as_os_str().to_str()) + .collect::>(); + ListFilesEntry { + path: components.join("/"), + size, + last_modified: last_modified.map(system_time_to_proto), + } + }) + }, + ) .collect::>(); Ok(Response::new(ListFilesResponse { files })) diff --git a/src/tonic/api/storage_read_api/tests.rs b/src/tonic/api/storage_read_api/tests.rs index d9e9756c95..3e754f7db5 100644 --- a/src/tonic/api/storage_read_api/tests.rs +++ b/src/tonic/api/storage_read_api/tests.rs @@ -267,18 +267,19 @@ async fn list_files_returns_paths_relative_to_shard_dir() { files.sort_by(|a, b| a.path.cmp(&b.path)); + for entry in &files { + assert!( + entry.last_modified.is_some(), + "local listing must carry a modification time", + ); + } + let paths_and_sizes: Vec<_> = files + .iter() + .map(|entry| (entry.path.as_str(), entry.size)) + .collect(); assert_eq!( - files, - vec![ - ListFilesEntry { - path: "index/chunk_1.bin".to_string(), - size: 3, - }, - ListFilesEntry { - path: "index/chunk_2.bin".to_string(), - size: 3, - }, - ] + paths_and_sizes, + [("index/chunk_1.bin", 3), ("index/chunk_2.bin", 3)] ); drop_service(service, storage_dir).await;