1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The Lance Authors
//! Trait for external manifest handler.
//!
//! This trait abstracts an external storage with put_if_not_exists semantics.
use std::sync::Arc;
use async_trait::async_trait;
use lance_core::{Error, Result};
use lance_io::object_store::{ObjectStore, ObjectStoreExt};
use log::warn;
use object_store::{path::Path, ObjectStore as OSObjectStore};
use snafu::{location, Location};
use super::{
current_manifest_path, make_staging_manifest_path, manifest_path, write_latest_manifest,
ManifestLocation, MANIFEST_EXTENSION,
};
use crate::format::{Index, Manifest};
use crate::io::commit::{CommitError, CommitHandler, ManifestWriter};
/// External manifest store
///
/// This trait abstracts an external storage for source of truth for manifests.
/// The storage is expected to remember (uri, version) -> manifest_path
/// and able to run transactions on the manifest_path.
///
/// This trait is called an **External** manifest store because the store is
/// expected to work in tandem with the object store. We are only leveraging
/// the external store for concurrent commit. Any manifest committed thru this
/// trait should ultimately be materialized in the object store.
/// For a visual explaination of the commit loop see
/// https://github.com/lancedb/lance/assets/12615154/b0822312-0826-432a-b554-3965f8d48d04
#[async_trait]
pub trait ExternalManifestStore: std::fmt::Debug + Send + Sync {
/// Get the manifest path for a given base_uri and version
async fn get(&self, base_uri: &str, version: u64) -> Result<String>;
/// Get the latest version of a dataset at the base_uri, and the path to the manifest.
/// The path is provided as an optimization. The path is deterministic based on
/// the version and the store should not customize it.
async fn get_latest_version(&self, base_uri: &str) -> Result<Option<(u64, String)>>;
/// Get the latest manifest location for a given base_uri.
///
/// By default, this calls get_latest_version. Impls should
/// override this method if they store both the location and size
/// of the latest manifest.
async fn get_latest_manifest_location(
&self,
base_uri: &str,
) -> Result<Option<ManifestLocation>> {
self.get_latest_version(base_uri).await.map(|res| {
res.map(|(version, uri)| ManifestLocation {
version,
path: Path::from(uri),
size: None,
})
})
}
/// Put the manifest path for a given base_uri and version, should fail if the version already exists
async fn put_if_not_exists(&self, base_uri: &str, version: u64, path: &str) -> Result<()>;
/// Put the manifest path for a given base_uri and version, should fail if the version **does not** already exist
async fn put_if_exists(&self, base_uri: &str, version: u64, path: &str) -> Result<()>;
}
/// External manifest commit handler
/// This handler is used to commit a manifest to an external store
/// for detailed design, see https://github.com/lancedb/lance/issues/1183
#[derive(Debug)]
pub struct ExternalManifestCommitHandler {
pub external_manifest_store: Arc<dyn ExternalManifestStore>,
}
#[async_trait]
impl CommitHandler for ExternalManifestCommitHandler {
async fn resolve_latest_location(
&self,
base_path: &Path,
object_store: &ObjectStore,
) -> std::result::Result<ManifestLocation, Error> {
let path = self.resolve_latest_version(base_path, object_store).await?;
Ok(ManifestLocation {
version: self
.resolve_latest_version_id(base_path, object_store)
.await?,
path,
size: None,
})
}
/// Get the latest version of a dataset at the path
async fn resolve_latest_version(
&self,
base_path: &Path,
object_store: &ObjectStore,
) -> std::result::Result<Path, Error> {
let version = self
.external_manifest_store
.get_latest_version(base_path.as_ref())
.await?;
match version {
Some((version, path)) => {
// The path is finalized, no need to check object store
if path.ends_with(&format!(".{MANIFEST_EXTENSION}")) {
return Ok(Path::parse(path)?);
}
// path is not finalized yet, we should try to finalize the path before loading
// if sync/finalize fails, return error
//
// step 1: copy path -> object_store_manifest_path
let object_store_manifest_path = manifest_path(base_path, version);
let manifest_path = Path::parse(path)?;
let staging = make_staging_manifest_path(&manifest_path)?;
// TODO: remove copy-rename once we upgrade object_store crate
object_store.copy(&manifest_path, &staging).await?;
object_store
.inner
.rename(&staging, &object_store_manifest_path)
.await?;
// step 2: write _latest.manifest
write_latest_manifest(&manifest_path, base_path, object_store.inner.as_ref())
.await?;
// step 3: update external store to finalize path
self.external_manifest_store
.put_if_exists(
base_path.as_ref(),
version,
object_store_manifest_path.as_ref(),
)
.await?;
Ok(object_store_manifest_path)
}
// Dataset not found in the external store, this could be because the dataset did not
// use external store for commit before. In this case, we search for the latest manifest
None => Ok(current_manifest_path(object_store, base_path).await?.path),
}
}
async fn resolve_latest_version_id(
&self,
base_path: &Path,
object_store: &ObjectStore,
) -> std::result::Result<u64, Error> {
let version = self
.external_manifest_store
.get_latest_version(base_path.as_ref())
.await?;
match version {
Some((version, _)) => Ok(version),
None => Ok(current_manifest_path(object_store, base_path)
.await?
.version),
}
}
async fn resolve_version(
&self,
base_path: &Path,
version: u64,
object_store: &dyn OSObjectStore,
) -> std::result::Result<Path, Error> {
let path_res = self
.external_manifest_store
.get(base_path.as_ref(), version)
.await;
let path = match path_res {
Ok(p) => p,
// not board external manifest yet, direct to object store
Err(Error::NotFound { .. }) => {
let path = manifest_path(base_path, version);
// if exist update external manifest store
if object_store.exists(&path).await? {
// best effort put, if it fails, it's okay
match self
.external_manifest_store
.put_if_not_exists(base_path.as_ref(), version, path.as_ref())
.await
{
Ok(_) => {}
Err(e) => {
warn!("could up update external manifest store during load, with error: {}", e);
}
}
} else {
return Err(Error::NotFound {
uri: path.to_string(),
location: location!(),
});
}
return Ok(manifest_path(base_path, version));
}
Err(e) => return Err(e),
};
// finalized path, just return
if path.ends_with(&format!(".{MANIFEST_EXTENSION}")) {
return Ok(Path::parse(path)?);
}
let manifest_path = manifest_path(base_path, version);
let staging_path = make_staging_manifest_path(&manifest_path)?;
// step1: try to materialize the manifest from external store to object store
// multiple writers could try to copy at the same time, this is okay
// as the content is immutable and copy is atomic
// We can't use `copy_if_not_exists` here because not all store supports it
object_store
.copy(&Path::parse(path)?, &staging_path)
.await?;
object_store.rename(&staging_path, &manifest_path).await?;
// finalize the external store
self.external_manifest_store
.put_if_exists(base_path.as_ref(), version, manifest_path.as_ref())
.await?;
Ok(manifest_path)
}
async fn commit(
&self,
manifest: &mut Manifest,
indices: Option<Vec<Index>>,
base_path: &Path,
object_store: &ObjectStore,
manifest_writer: ManifestWriter,
) -> std::result::Result<(), CommitError> {
// path we get here is the path to the manifest we want to write
// use object_store.base_path.as_ref() for getting the root of the dataset
// step 1: Write the manifest we want to commit to object store with a temporary name
let path = manifest_path(base_path, manifest.version);
let staging_path = make_staging_manifest_path(&path)?;
manifest_writer(object_store, manifest, indices, &staging_path).await?;
// step 2 & 3: Try to commit this version to external store, return err on failure
// TODO: add logic to clean up orphaned staged manifests, the ones that failed to commit to external store
// https://github.com/lancedb/lance/issues/1201
self.external_manifest_store
.put_if_not_exists(base_path.as_ref(), manifest.version, staging_path.as_ref())
.await
.map_err(|_| CommitError::CommitConflict {})?;
// step 4: copy the manifest to the final location
object_store.inner.copy(
&staging_path,
&path,
).await.map_err(|e| CommitError::OtherError(
Error::io(
format!("commit to external store is successful, but could not copy manifest to object store, with error: {}.", e),
location!(),
)
))?;
// update the _latest.manifest pointer
write_latest_manifest(&path, base_path, &object_store.inner).await?;
// step 5: flip the external store to point to the final location
self.external_manifest_store
.put_if_exists(base_path.as_ref(), manifest.version, path.as_ref())
.await?;
Ok(())
}
}