Set-time checking

icechunk/src/dataset.rs

@@ -8,7 +8,11 @@ use std::{
 
 use crate::storage::virtual_ref::VirtualChunkResolver;
 pub use crate::{
-    format::{manifest::ChunkPayload, snapshot::ZarrArrayMetadata, ChunkIndices, Path},
+    format::{
+        manifest::{ChunkPayload, VirtualChunkLocation},
+        snapshot::ZarrArrayMetadata,
+        ChunkIndices, Path,
+    },
     metadata::{
         ArrayShape, ChunkKeyEncoding, ChunkShape, Codec, DataType, DimensionName,
         DimensionNames, FillValue, StorageTransformer, UserAttributes,
@@ -150,7 +154,6 @@ impl ChangeSet {
         coord: ChunkIndices,
         data: Option<ChunkPayload>,
     ) {
-        // FIXME: normalize virtual chunk reference paths here
         // this implementation makes delete idempotent
         // it allows deleting a deleted chunk by repeatedly setting None.
         self.set_chunks
@@ -1146,7 +1149,9 @@ mod tests {
     use std::{error::Error, num::NonZeroU64, path::PathBuf};
 
     use crate::{
-        format::manifest::{ChunkInfo, VirtualChunkLocation, VirtualChunkRef},
+        format::manifest::{
+            ChunkInfo, VirtualChunkLocation, VirtualChunkRef, VirtualReferenceError,
+        },
         metadata::{
             ChunkKeyEncoding, ChunkShape, Codec, DataType, FillValue, StorageTransformer,
         },
@@ -1326,22 +1331,34 @@ mod tests {
             dimension_names: None,
         };
         let payload1 = ChunkPayload::Virtual(VirtualChunkRef {
-            location: VirtualChunkLocation::Absolute(format!(
-                "s3://testbucket/{}",
+            location: VirtualChunkLocation::from_absolute_path(&format!(
+                // intentional extra '/'
+                "s3://testbucket///{}",
                 path1
-            )),
+            ))?,
             offset: 0,
             length: 5,
         });
         let payload2 = ChunkPayload::Virtual(VirtualChunkRef {
-            location: VirtualChunkLocation::Absolute(format!(
+            location: VirtualChunkLocation::from_absolute_path(&format!(
                 "s3://testbucket/{}",
                 path2
-            )),
+            ))?,
             offset: 1,
             length: 5,
         });
 
+        // bad relative chunk location
+        assert!(matches!(
+            VirtualChunkLocation::from_absolute_path("abcdef"),
+            Err(VirtualReferenceError::CannotParseUrl(_)),
+        ));
+        // extra / prevents bucket name detection
+        assert!(matches!(
+            VirtualChunkLocation::from_absolute_path("s3:///foo/path"),
+            Err(VirtualReferenceError::CannotParseBucketName(_)),
+        ));
+
         let new_array_path: PathBuf = "/array".to_string().into();
         ds.add_array(new_array_path.clone(), zarr_meta.clone()).await?;
 

icechunk/src/format/manifest.rs

@@ -1,4 +1,6 @@
+use itertools::Itertools;
 use std::{collections::BTreeMap, ops::Bound, sync::Arc};
+use thiserror::Error;
 
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
@@ -15,12 +17,53 @@ pub struct ManifestRef {
     pub extents: ManifestExtents,
 }
 
+#[derive(Debug, Error)]
+pub enum VirtualReferenceError {
+    #[error("error parsing virtual ref URL {0}")]
+    CannotParseUrl(#[from] url::ParseError),
+    #[error("virtual reference has no path segments {0}")]
+    NoPathSegments(String),
+    #[error("unsupported scheme for virtual chunk refs: {0}")]
+    UnsupportedScheme(String),
+    #[error("error parsing bucket name from virtual ref URL {0}")]
+    CannotParseBucketName(String),
+    #[error("error parsing path using object_store {0}")]
+    CannotParsePath(#[from] ::object_store::path::Error),
+}
 #[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub enum VirtualChunkLocation {
     Absolute(String),
     // Relative(prefix_id, String)
 }
 
+impl VirtualChunkLocation {
+    pub fn from_absolute_path(
+        path: &str,
+    ) -> Result<VirtualChunkLocation, VirtualReferenceError> {
+        // make sure we can parse the provided URL before creating the enum
+        // TODO: consider other validation here.
+        let url = url::Url::parse(path)?;
+        let new_path: String = url
+            .path_segments()
+            .ok_or(VirtualReferenceError::NoPathSegments(path.into()))?
+            // strip empty segments here, object_store cannot handle them.
+            .filter(|x| !x.is_empty())
+            .join("/");
+
+        if !url.has_host() {
+            Err(VirtualReferenceError::CannotParseBucketName(path.into()))
+        } else {
+            #[allow(clippy::expect_used)]
+            Ok(VirtualChunkLocation::Absolute(format!(
+                "{}://{}/{}",
+                url.scheme(),
+                url.host().expect("cannot be absent."),
+                new_path,
+            )))
+        }
+    }
+}
+
 #[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub struct VirtualChunkRef {
     pub location: VirtualChunkLocation,

icechunk/src/storage/mod.rs

@@ -18,8 +18,10 @@ pub use caching::MemCachingStorage;
 pub use object_store::ObjectStorage;
 
 use crate::format::{
-    attributes::AttributesTable, manifest::Manifest, snapshot::Snapshot, ByteRange,
-    ObjectId, Path,
+    attributes::AttributesTable,
+    manifest::{Manifest, VirtualReferenceError},
+    snapshot::Snapshot,
+    ByteRange, ObjectId, Path,
 };
 
 #[derive(Debug, Error)]
@@ -38,14 +40,8 @@ pub enum StorageError {
     RefAlreadyExists(String),
     #[error("ref not found: {0}")]
     RefNotFound(String),
-    #[error("error parsing virtual ref URL {0}")]
-    VirtualUrlError(#[from] url::ParseError),
-    #[error("error parsing bucket name from virtual ref URL {0}")]
-    VirtualBucketParseError(String),
-    #[error("error parsing path using object_store {0}")]
-    VirtualPathParseError(#[from] ::object_store::path::Error),
-    #[error("unsupported scheme for virtual chunk refs: {0}")]
-    UnsupportedScheme(String),
+    #[error("error parsing virtual reference {0}")]
+    VirtualReferenceError(#[from] VirtualReferenceError),
     #[error("generic storage error: {0}")]
     OtherError(#[from] Arc<dyn std::error::Error + Sync + Send>),
     #[error("unknown storage error: {0}")]

icechunk/src/storage/object_store.rs

@@ -6,7 +6,7 @@ use std::{
 
 use crate::format::{
     attributes::AttributesTable,
-    manifest::{Manifest, VirtualChunkLocation},
+    manifest::{Manifest, VirtualChunkLocation, VirtualReferenceError},
     snapshot::Snapshot,
     ByteRange, ObjectId,
 };
@@ -381,14 +381,16 @@ impl VirtualChunkResolver for ObjectStoreVirtualChunkResolver {
         range: &ByteRange,
     ) -> StorageResult<Bytes> {
         let VirtualChunkLocation::Absolute(location) = location;
-        let parsed = url::Url::parse(location)?;
+        let parsed = url::Url::parse(location)
+            .map_err(VirtualReferenceError::CannotParseUrl)?;
         let bucket_name = parsed
             .host_str()
-            .ok_or(StorageError::VirtualBucketParseError(
+            .ok_or(VirtualReferenceError::CannotParseBucketName(
                 "error parsing bucket name".into(),
             ))?
             .to_string();
-        let path = ObjectPath::parse(parsed.path())?;
+        let path = ObjectPath::parse(parsed.path())
+            .map_err(VirtualReferenceError::CannotParsePath)?;
         let scheme = parsed.scheme();
         let cache_key = StoreCacheKey(scheme.into(), bucket_name);
 
@@ -400,7 +402,9 @@ impl VirtualChunkResolver for ObjectStoreVirtualChunkResolver {
             false => {
                 let builder = match scheme {
                     "s3" => AmazonS3Builder::from_env(),
-                    _ => Err(StorageError::UnsupportedScheme(scheme.to_string()))?,
+                    _ => {
+                        Err(VirtualReferenceError::UnsupportedScheme(scheme.to_string()))?
+                    }
                 };
                 let new_store = Arc::new(builder.with_bucket_name(&cache_key.1).build()?);
                 {

icechunk/src/storage/virtual_ref.rs

@@ -0,0 +1,13 @@
+use crate::format::{manifest::VirtualChunkLocation, ByteRange};
+use crate::storage::StorageResult;
+use async_trait::async_trait;
+use bytes::Bytes;
+
+#[async_trait]
+pub trait VirtualChunkResolver {
+    async fn fetch_chunk(
+        &self,
+        location: &VirtualChunkLocation,
+        range: &ByteRange,
+    ) -> StorageResult<Bytes>;
+}