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
// Copyright 2023 Lance Developers.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! 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 log::warn;
use object_store::{path::Path, ObjectStore};
use snafu::{location, Location};
use super::{
current_manifest_path, make_staging_manifest_path, manifest_path, write_latest_manifest,
MANIFEST_EXTENSION,
};
use crate::format::{Index, Manifest};
use crate::io::{
commit::{parse_version_from_path, CommitError, CommitHandler, ManifestWriter},
object_store::ObjectStoreExt,
};
use crate::{Error, Result};
/// External manifest store
///
/// This trait abstracts an external storage for source of truth for manifests.
/// The storge 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)>>;
/// 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 {
/// Get the latest version of a dataset at the path
async fn resolve_latest_version(
&self,
base_path: &Path,
object_store: &dyn 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
.rename(&staging, &object_store_manifest_path)
.await?;
// step 2: write _latest.manifest
write_latest_manifest(&manifest_path, base_path, object_store).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 => current_manifest_path(object_store, base_path).await,
}
}
async fn resolve_latest_version_id(
&self,
base_path: &Path,
object_store: &dyn ObjectStore,
) -> std::result::Result<u64, crate::Error> {
let version = self
.external_manifest_store
.get_latest_version(base_path.as_ref())
.await?;
match version {
Some((version, _)) => Ok(version),
None => parse_version_from_path(¤t_manifest_path(object_store, base_path).await?),
}
}
async fn resolve_version(
&self,
base_path: &Path,
version: u64,
object_store: &dyn ObjectStore,
) -> 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: &dyn 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.copy(
&staging_path,
&path,
).await.map_err(|e| CommitError::OtherError(
Error::IO {
message: format!("commit to external store is successful, but could not copy manifest to object store, with error: {}.", e),
location: location!(),
}
))?;
// update the _latest.manifest pointer
write_latest_manifest(&path, base_path, object_store).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(())
}
}