1use async_trait::async_trait;
5use bytes::Bytes;
6use chrono::prelude::*;
7use lance_core::deepsize::DeepSizeOf;
8use lance_file::datatypes::{Fields, FieldsWithMeta};
9use lance_file::version::{ConcreteFileVersion, stable_file_version};
10use lance_file::versions::v1::{
11 encoding::populate_schema_dictionaries, reader::FileReader as V1FileReader,
12};
13use lance_io::traits::{ProtoStruct, Reader};
14use object_store::path::Path;
15use prost::Message;
16use prost_types::Timestamp;
17use std::collections::{BTreeMap, HashMap};
18use std::ops::Range;
19use std::sync::Arc;
20
21use super::{Fragment, InlineRowIds, RowIdMeta};
22use crate::feature_flags::{FLAG_COVERED_INDEX_METADATA, STICKY_PAIRED_FLAGS};
23use crate::feature_flags::{FLAG_STABLE_ROW_IDS, has_deprecated_v2_feature_flag};
24use crate::format::fragment::DataFileFieldInterner;
25use crate::format::pb;
26use lance_core::cache::LanceCache;
27use lance_core::datatypes::Schema;
28use lance_core::{Error, Result};
29use lance_io::object_store::{ObjectStore, ObjectStoreRegistry};
30use lance_io::utils::read_struct;
31
32#[derive(Debug, Clone, PartialEq, DeepSizeOf)]
39pub struct Manifest {
40 pub schema: Schema,
42
43 pub version: u64,
45
46 pub branch: Option<String>,
48
49 pub writer_version: Option<WriterVersion>,
51
52 pub fragments: Arc<Vec<Fragment>>,
57
58 pub version_aux_data: usize,
60
61 pub index_section: Option<usize>,
63
64 pub timestamp_nanos: u128,
66
67 pub tag: Option<String>,
69
70 pub reader_feature_flags: u64,
72
73 pub writer_feature_flags: u64,
75
76 pub max_fragment_id: Option<u32>,
79
80 pub transaction_file: Option<String>,
82
83 pub transaction_section: Option<usize>,
85
86 fragment_offsets: Vec<usize>,
89
90 pub next_row_id: u64,
92
93 pub data_storage_format: DataStorageFormat,
95
96 pub config: HashMap<String, String>,
98
99 pub table_metadata: HashMap<String, String>,
105
106 pub base_paths: HashMap<u32, BasePath>,
108}
109
110pub const DETACHED_VERSION_MASK: u64 = 0x8000_0000_0000_0000;
112
113pub fn is_detached_version(version: u64) -> bool {
114 version & DETACHED_VERSION_MASK != 0
115}
116
117fn compute_fragment_offsets(fragments: &[Fragment]) -> Vec<usize> {
118 fragments
119 .iter()
120 .map(|f| f.num_rows().unwrap_or_default())
121 .chain([0]) .scan(0_usize, |offset, len| {
123 let start = *offset;
124 *offset += len;
125 Some(start)
126 })
127 .collect()
128}
129
130#[derive(Default)]
131pub struct ManifestSummary {
132 pub total_fragments: u64,
133 pub total_data_files: u64,
134 pub total_files_size: u64,
135 pub total_deletion_files: u64,
136 pub total_data_file_rows: u64,
137 pub total_deletion_file_rows: u64,
138 pub total_rows: u64,
139}
140
141impl From<ManifestSummary> for BTreeMap<String, String> {
142 fn from(summary: ManifestSummary) -> Self {
143 let mut stats_map = Self::new();
144 stats_map.insert(
145 "total_fragments".to_string(),
146 summary.total_fragments.to_string(),
147 );
148 stats_map.insert(
149 "total_data_files".to_string(),
150 summary.total_data_files.to_string(),
151 );
152 stats_map.insert(
153 "total_files_size".to_string(),
154 summary.total_files_size.to_string(),
155 );
156 stats_map.insert(
157 "total_deletion_files".to_string(),
158 summary.total_deletion_files.to_string(),
159 );
160 stats_map.insert(
161 "total_data_file_rows".to_string(),
162 summary.total_data_file_rows.to_string(),
163 );
164 stats_map.insert(
165 "total_deletion_file_rows".to_string(),
166 summary.total_deletion_file_rows.to_string(),
167 );
168 stats_map.insert("total_rows".to_string(), summary.total_rows.to_string());
169 stats_map
170 }
171}
172
173impl Manifest {
174 pub fn new(
175 schema: Schema,
176 fragments: Arc<Vec<Fragment>>,
177 data_storage_format: DataStorageFormat,
178 base_paths: HashMap<u32, BasePath>,
179 ) -> Self {
180 let fragment_offsets = compute_fragment_offsets(&fragments);
181
182 Self {
183 schema,
184 version: 1,
185 branch: None,
186 writer_version: Some(WriterVersion::default()),
187 fragments,
188 version_aux_data: 0,
189 index_section: None,
190 timestamp_nanos: 0,
191 tag: None,
192 reader_feature_flags: 0, writer_feature_flags: 0, max_fragment_id: None,
195 transaction_file: None,
196 transaction_section: None,
197 fragment_offsets,
198 next_row_id: 0,
199 data_storage_format,
200 config: HashMap::new(),
201 table_metadata: HashMap::new(),
202 base_paths,
203 }
204 }
205
206 pub fn new_from_previous(
207 previous: &Self,
208 schema: Schema,
209 fragments: Arc<Vec<Fragment>>,
210 ) -> Self {
211 let fragment_offsets = compute_fragment_offsets(&fragments);
212
213 Self {
214 schema,
215 version: previous.version + 1,
216 branch: previous.branch.clone(),
217 writer_version: Some(WriterVersion::default()),
218 fragments,
219 version_aux_data: 0,
220 index_section: None, timestamp_nanos: 0, tag: None,
223 reader_feature_flags: previous.reader_feature_flags & STICKY_PAIRED_FLAGS,
224 writer_feature_flags: previous.writer_feature_flags & STICKY_PAIRED_FLAGS,
225 max_fragment_id: previous.max_fragment_id,
226 transaction_file: None,
227 transaction_section: None,
228 fragment_offsets,
229 next_row_id: previous.next_row_id,
230 data_storage_format: previous.data_storage_format.clone(),
231 config: previous.config.clone(),
232 table_metadata: previous.table_metadata.clone(),
233 base_paths: previous.base_paths.clone(),
234 }
235 }
236
237 pub fn shallow_clone(
242 &self,
243 ref_name: Option<String>,
244 ref_path: String,
245 ref_base_id: u32,
246 branch_name: Option<String>,
247 transaction_file: String,
248 ) -> Self {
249 let cloned_fragments = self
250 .fragments
251 .as_ref()
252 .iter()
253 .map(|fragment| {
254 let mut cloned_fragment = fragment.clone();
255 for file in cloned_fragment.referenced_lance_files_mut() {
256 if file.base_id.is_none() {
257 file.base_id = Some(ref_base_id);
258 }
259 }
260
261 if let Some(deletion) = &mut cloned_fragment.deletion_file
262 && deletion.base_id.is_none()
263 {
264 deletion.base_id = Some(ref_base_id);
265 }
266 cloned_fragment
267 })
268 .collect::<Vec<_>>();
269
270 Self {
271 schema: self.schema.clone(),
272 version: self.version,
273 branch: branch_name,
274 writer_version: self.writer_version.clone(),
275 fragments: Arc::new(cloned_fragments),
276 version_aux_data: self.version_aux_data,
277 index_section: None, timestamp_nanos: self.timestamp_nanos,
279 tag: None,
280 reader_feature_flags: self.reader_feature_flags
290 & (FLAG_COVERED_INDEX_METADATA | STICKY_PAIRED_FLAGS),
291 writer_feature_flags: self.writer_feature_flags
292 & (FLAG_COVERED_INDEX_METADATA | STICKY_PAIRED_FLAGS),
293 max_fragment_id: self.max_fragment_id,
294 transaction_file: Some(transaction_file),
295 transaction_section: None,
296 fragment_offsets: self.fragment_offsets.clone(),
297 next_row_id: self.next_row_id,
298 data_storage_format: self.data_storage_format.clone(),
299 config: self.config.clone(),
300 base_paths: {
301 let mut base_paths = self.base_paths.clone();
302 let base_path = BasePath::new(ref_base_id, ref_path, ref_name, true);
303 base_paths.insert(ref_base_id, base_path);
304 base_paths
305 },
306 table_metadata: self.table_metadata.clone(),
307 }
308 }
309
310 pub fn timestamp(&self) -> DateTime<Utc> {
312 let nanos = self.timestamp_nanos % 1_000_000_000;
313 let seconds = ((self.timestamp_nanos - nanos) / 1_000_000_000) as i64;
314 Utc.from_utc_datetime(
315 &DateTime::from_timestamp(seconds, nanos as u32)
316 .unwrap_or_default()
317 .naive_utc(),
318 )
319 }
320
321 pub fn set_timestamp(&mut self, nanos: u128) {
323 self.timestamp_nanos = nanos;
324 }
325
326 pub fn config_mut(&mut self) -> &mut HashMap<String, String> {
328 &mut self.config
329 }
330
331 pub fn table_metadata_mut(&mut self) -> &mut HashMap<String, String> {
333 &mut self.table_metadata
334 }
335
336 pub fn schema_metadata_mut(&mut self) -> &mut HashMap<String, String> {
338 &mut self.schema.metadata
339 }
340
341 pub fn field_metadata_mut(&mut self, field_id: i32) -> Option<&mut HashMap<String, String>> {
345 self.schema
346 .field_by_id_mut(field_id)
347 .map(|field| &mut field.metadata)
348 }
349
350 #[deprecated(note = "Use config_mut() for direct access to config HashMap")]
352 pub fn update_config(&mut self, upsert_values: impl IntoIterator<Item = (String, String)>) {
353 self.config.extend(upsert_values);
354 }
355
356 #[deprecated(note = "Use config_mut() for direct access to config HashMap")]
358 pub fn delete_config_keys(&mut self, delete_keys: &[&str]) {
359 self.config
360 .retain(|key, _| !delete_keys.contains(&key.as_str()));
361 }
362
363 #[deprecated(note = "Use schema_metadata_mut() for direct access to schema metadata HashMap")]
365 pub fn replace_schema_metadata(&mut self, new_metadata: HashMap<String, String>) {
366 self.schema.metadata = new_metadata;
367 }
368
369 #[deprecated(
373 note = "Use field_metadata_mut(field_id) for direct access to field metadata HashMap"
374 )]
375 pub fn replace_field_metadata(
376 &mut self,
377 field_id: i32,
378 new_metadata: HashMap<String, String>,
379 ) -> Result<()> {
380 if let Some(field) = self.schema.field_by_id_mut(field_id) {
381 field.metadata = new_metadata;
382 Ok(())
383 } else {
384 Err(Error::invalid_input(format!(
385 "Field with id {} does not exist for replace_field_metadata",
386 field_id
387 )))
388 }
389 }
390
391 pub fn detach_sparse_inline_row_ids(&mut self, buffer_len: usize) {
401 let inline_bytes: usize = self
402 .fragments
403 .iter()
404 .filter_map(|fragment| match &fragment.row_id_meta {
405 Some(RowIdMeta::Inline(data)) => Some(data.len()),
406 _ => None,
407 })
408 .sum();
409 if inline_bytes == 0 || inline_bytes.saturating_mul(4) >= buffer_len {
411 return;
412 }
413 for fragment in Arc::make_mut(&mut self.fragments) {
414 if let Some(RowIdMeta::Inline(data)) = &fragment.row_id_meta {
415 let copied = InlineRowIds::from(Bytes::copy_from_slice(data));
416 fragment.row_id_meta = Some(RowIdMeta::Inline(copied));
417 }
418 }
419 }
420
421 pub fn update_max_fragment_id(&mut self) {
423 if self.fragments.is_empty() {
425 return;
426 }
427
428 let max_fragment_id = self
429 .fragments
430 .iter()
431 .map(|f| f.id)
432 .max()
433 .unwrap() .try_into()
435 .unwrap();
436
437 match self.max_fragment_id {
438 None => {
439 self.max_fragment_id = Some(max_fragment_id);
441 }
442 Some(current_max) => {
443 if max_fragment_id > current_max {
446 self.max_fragment_id = Some(max_fragment_id);
447 }
448 }
449 }
450 }
451
452 pub fn max_fragment_id(&self) -> Option<u64> {
457 if let Some(max_id) = self.max_fragment_id {
458 Some(max_id.into())
460 } else {
461 self.fragments.iter().map(|f| f.id).max()
463 }
464 }
465
466 pub fn max_field_id(&self) -> i32 {
472 let schema_max_id = self.schema.max_field_id().unwrap_or(-1);
473 let fragment_max_id = self
474 .fragments
475 .iter()
476 .flat_map(|fragment| {
477 fragment
478 .referenced_lance_files()
479 .flat_map(|file| file.fields.iter())
480 })
481 .copied()
482 .max()
483 .unwrap_or(-1);
484 schema_max_id.max(fragment_max_id)
485 }
486
487 pub fn fragments_since(&self, since: &Self) -> Result<Vec<Fragment>> {
490 if since.version >= self.version {
491 return Err(Error::invalid_input(format!(
492 "fragments_since: given version {} is newer than manifest version {}",
493 since.version, self.version
494 )));
495 }
496 let start = since.max_fragment_id();
497 Ok(self
498 .fragments
499 .iter()
500 .filter(|&f| start.map(|s| f.id > s).unwrap_or(true))
501 .cloned()
502 .collect())
503 }
504
505 pub fn fragments_by_offset_range(&self, range: Range<usize>) -> Vec<(usize, &Fragment)> {
521 let start = range.start;
522 let end = range.end;
523 let idx = self
524 .fragment_offsets
525 .binary_search(&start)
526 .unwrap_or_else(|idx| idx - 1);
527
528 let mut fragments = vec![];
529 for i in idx..self.fragments.len() {
530 if self.fragment_offsets[i] >= end
531 || self.fragment_offsets[i] + self.fragments[i].num_rows().unwrap_or_default()
532 <= start
533 {
534 break;
535 }
536 fragments.push((self.fragment_offsets[i], &self.fragments[i]));
537 }
538
539 fragments
540 }
541
542 pub fn uses_stable_row_ids(&self) -> bool {
544 self.reader_feature_flags & FLAG_STABLE_ROW_IDS != 0
545 }
546
547 pub fn serialized(&self) -> Vec<u8> {
550 let pb_manifest: pb::Manifest = self.into();
551 pb_manifest.encode_to_vec()
552 }
553
554 pub fn summary(&self) -> ManifestSummary {
565 let mut summary =
567 self.fragments
568 .iter()
569 .fold(ManifestSummary::default(), |mut summary, f| {
570 summary.total_data_files += f.files.len() as u64;
572 if let Some(num_rows) = f.num_rows() {
574 summary.total_rows += num_rows as u64;
575 }
576 for data_file in &f.files {
578 if let Some(size_bytes) = data_file.file_size_bytes.get() {
579 summary.total_files_size += size_bytes.get();
580 }
581 }
582 if f.deletion_file.is_some() {
584 summary.total_deletion_files += 1;
585 }
586 if let Some(deletion_file) = &f.deletion_file
588 && let Some(num_deleted) = deletion_file.num_deleted_rows
589 {
590 summary.total_deletion_file_rows += num_deleted as u64;
591 }
592 summary
593 });
594 summary.total_fragments = self.fragments.len() as u64;
595 summary.total_data_file_rows = summary.total_rows + summary.total_deletion_file_rows;
596
597 summary
598 }
599}
600
601pub async fn populate_manifest_schema_dictionaries(
620 manifest: &mut Manifest,
621 reader: &dyn Reader,
622) -> Result<()> {
623 match manifest.data_storage_format.version {
624 ConcreteFileVersion::V1 => {
625 populate_schema_dictionaries(&mut manifest.schema, reader).await?;
626 }
627 ConcreteFileVersion::V2_0
628 | ConcreteFileVersion::V2_1
629 | ConcreteFileVersion::V2_2
630 | ConcreteFileVersion::V2_3 => {}
631 }
632 Ok(())
633}
634
635#[derive(Debug, Clone, PartialEq)]
636pub struct BasePath {
637 pub id: u32,
638 pub name: Option<String>,
639 pub is_dataset_root: bool,
640 pub path: String,
642}
643
644impl BasePath {
645 pub fn new(id: u32, path: String, name: Option<String>, is_dataset_root: bool) -> Self {
654 Self {
655 id,
656 name,
657 is_dataset_root,
658 path,
659 }
660 }
661
662 pub fn extract_path(&self, registry: Arc<ObjectStoreRegistry>) -> Result<Path> {
666 ObjectStore::extract_path_from_uri(registry, &self.path)
667 }
668}
669
670impl DeepSizeOf for BasePath {
671 fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize {
672 self.name.deep_size_of_children(context)
673 + self.path.deep_size_of_children(context) * 2
674 + size_of::<bool>()
675 }
676}
677
678#[derive(Debug, Clone, PartialEq, DeepSizeOf)]
679pub struct WriterVersion {
680 pub library: String,
681 pub version: String,
682 pub prerelease: Option<String>,
683 pub build_metadata: Option<String>,
684}
685
686#[derive(Debug, Clone, PartialEq, DeepSizeOf)]
687pub struct DataStorageFormat {
688 pub file_format: String,
689 pub version: ConcreteFileVersion,
690}
691
692const LANCE_FORMAT_NAME: &str = "lance";
693
694impl DataStorageFormat {
695 pub fn new(version: ConcreteFileVersion) -> Self {
696 Self {
697 file_format: LANCE_FORMAT_NAME.to_string(),
698 version,
699 }
700 }
701
702 pub fn lance_file_format(&self) -> ConcreteFileVersion {
704 self.version
705 }
706}
707
708impl Default for DataStorageFormat {
709 fn default() -> Self {
710 Self::new(stable_file_version())
711 }
712}
713
714impl TryFrom<pb::manifest::DataStorageFormat> for DataStorageFormat {
715 type Error = Error;
716
717 fn try_from(pb: pb::manifest::DataStorageFormat) -> Result<Self> {
718 Ok(Self {
719 file_format: pb.file_format,
720 version: ConcreteFileVersion::from_manifest_string(&pb.version)?,
721 })
722 }
723}
724
725#[derive(Debug, Clone)]
732pub struct ManifestBuildConfig {
733 pub auto_set_feature_flags: bool,
736 pub timestamp_nanos: u128,
738 pub use_stable_row_ids: bool,
741 pub use_legacy_format: Option<bool>,
744 pub storage_format: Option<DataStorageFormat>,
747 pub disable_transaction_file: bool,
749 pub migration_next_row_id: Option<u64>,
753 pub spilled_row_lineage: std::sync::Arc<crate::rowids::version::SpilledRowLineage>,
760}
761
762#[derive(Debug, Clone, Copy, PartialEq, Eq)]
763pub enum VersionPart {
764 Major,
765 Minor,
766 Patch,
767}
768
769fn bump_version(version: &mut semver::Version, part: VersionPart) {
770 match part {
771 VersionPart::Major => {
772 version.major += 1;
773 version.minor = 0;
774 version.patch = 0;
775 }
776 VersionPart::Minor => {
777 version.minor += 1;
778 version.patch = 0;
779 }
780 VersionPart::Patch => {
781 version.patch += 1;
782 }
783 }
784}
785
786impl WriterVersion {
787 fn split_version(full_version: &str) -> Option<(String, Option<String>, Option<String>)> {
797 let mut parsed = semver::Version::parse(full_version).ok()?;
798
799 let prerelease = if parsed.pre.is_empty() {
800 None
801 } else {
802 Some(parsed.pre.to_string())
803 };
804
805 let build_metadata = if parsed.build.is_empty() {
806 None
807 } else {
808 Some(parsed.build.to_string())
809 };
810
811 parsed.pre = semver::Prerelease::EMPTY;
813 parsed.build = semver::BuildMetadata::EMPTY;
814 Some((parsed.to_string(), prerelease, build_metadata))
815 }
816
817 #[deprecated(note = "Use `lance_lib_version()` instead")]
820 pub fn semver(&self) -> Option<(u32, u32, u32, Option<&str>)> {
821 let (version_part, tag) = if let Some(dash_idx) = self.version.find('-') {
823 (
824 &self.version[..dash_idx],
825 Some(&self.version[dash_idx + 1..]),
826 )
827 } else {
828 (self.version.as_str(), None)
829 };
830
831 let mut parts = version_part.split('.');
832 let major = parts.next().unwrap_or("0").parse().ok()?;
833 let minor = parts.next().unwrap_or("0").parse().ok()?;
834 let patch = parts.next().unwrap_or("0").parse().ok()?;
835
836 Some((major, minor, patch, tag))
837 }
838
839 pub fn lance_lib_version(&self) -> Option<semver::Version> {
847 if self.library != "lance" {
848 return None;
849 }
850
851 let mut version = semver::Version::parse(&self.version).ok()?;
852
853 if let Some(ref prerelease) = self.prerelease {
854 version.pre = semver::Prerelease::new(prerelease).ok()?;
855 }
856
857 if let Some(ref build_metadata) = self.build_metadata {
858 version.build = semver::BuildMetadata::new(build_metadata).ok()?;
859 }
860
861 Some(version)
862 }
863
864 #[deprecated(
865 note = "Use `lance_lib_version()` instead, which safely checks the library field and returns Option"
866 )]
867 #[allow(deprecated)]
868 pub fn semver_or_panic(&self) -> (u32, u32, u32, Option<&str>) {
869 self.semver()
870 .unwrap_or_else(|| panic!("Invalid writer version: {}", self.version))
871 }
872
873 #[deprecated(note = "Use `lance_lib_version()` and its `older_than` method instead.")]
879 pub fn older_than(&self, major: u32, minor: u32, patch: u32) -> bool {
880 let version = self
881 .lance_lib_version()
882 .expect("Not lance library or invalid version");
883 let other = semver::Version {
884 major: major.into(),
885 minor: minor.into(),
886 patch: patch.into(),
887 pre: semver::Prerelease::EMPTY,
888 build: semver::BuildMetadata::EMPTY,
889 };
890 version < other
891 }
892
893 #[deprecated(note = "This is meant for testing and will be made private in future version.")]
894 pub fn bump(&self, part: VersionPart, keep_tag: bool) -> Self {
895 let mut version = self.lance_lib_version().expect("Should be lance version");
896 bump_version(&mut version, part);
897 if !keep_tag {
898 version.pre = semver::Prerelease::EMPTY;
899 }
900 let (clean_version, prerelease, build_metadata) = Self::split_version(&version.to_string())
901 .expect("Bumped version should be valid semver");
902 Self {
903 library: self.library.clone(),
904 version: clean_version,
905 prerelease,
906 build_metadata,
907 }
908 }
909}
910
911impl Default for WriterVersion {
912 #[cfg(not(test))]
913 fn default() -> Self {
914 let full_version = env!("CARGO_PKG_VERSION");
915 let (version, prerelease, build_metadata) =
916 Self::split_version(full_version).expect("CARGO_PKG_VERSION should be valid semver");
917 Self {
918 library: "lance".to_string(),
919 version,
920 prerelease,
921 build_metadata,
922 }
923 }
924
925 #[cfg(test)]
927 #[allow(deprecated)]
928 fn default() -> Self {
929 let full_version = env!("CARGO_PKG_VERSION");
930 let (version, prerelease, build_metadata) =
931 Self::split_version(full_version).expect("CARGO_PKG_VERSION should be valid semver");
932 Self {
933 library: "lance".to_string(),
934 version,
935 prerelease,
936 build_metadata,
937 }
938 .bump(VersionPart::Patch, true)
939 }
940}
941
942impl ProtoStruct for Manifest {
943 type Proto = pb::Manifest;
944}
945
946impl From<pb::BasePath> for BasePath {
947 fn from(p: pb::BasePath) -> Self {
948 Self::new(p.id, p.path, p.name, p.is_dataset_root)
949 }
950}
951
952impl From<BasePath> for pb::BasePath {
953 fn from(p: BasePath) -> Self {
954 Self {
955 id: p.id,
956 name: p.name,
957 is_dataset_root: p.is_dataset_root,
958 path: p.path,
959 }
960 }
961}
962
963impl TryFrom<pb::Manifest> for Manifest {
964 type Error = Error;
965
966 fn try_from(p: pb::Manifest) -> Result<Self> {
967 let timestamp_nanos = p.timestamp.map(|ts| {
968 let sec = ts.seconds as u128 * 1e9 as u128;
969 let nanos = ts.nanos as u128;
970 sec + nanos
971 });
972 let writer_version = match p.writer_version {
974 Some(pb::manifest::WriterVersion {
975 library,
976 version,
977 prerelease,
978 build_metadata,
979 }) => Some(WriterVersion {
980 library,
981 version,
982 prerelease,
983 build_metadata,
984 }),
985 _ => None,
986 };
987 let mut interner = DataFileFieldInterner::default();
988 let fragments = Arc::new(
989 p.fragments
990 .into_iter()
991 .map(|f| interner.intern_fragment(f))
992 .collect::<Result<Vec<_>>>()?,
993 );
994 let fragment_offsets = compute_fragment_offsets(fragments.as_slice());
995 let fields_with_meta = FieldsWithMeta {
996 fields: Fields(p.fields),
997 metadata: p.schema_metadata,
998 };
999
1000 if FLAG_STABLE_ROW_IDS & p.reader_feature_flags != 0
1001 && !fragments.iter().all(|frag| frag.row_id_meta.is_some())
1002 {
1003 return Err(Error::internal("All fragments must have row ids"));
1004 }
1005
1006 let data_storage_format = match p.data_format {
1007 None => {
1008 if let Some(inferred_version) = Fragment::try_infer_version(fragments.as_ref())? {
1009 DataStorageFormat::new(inferred_version)
1011 } else {
1012 if has_deprecated_v2_feature_flag(p.writer_feature_flags) {
1014 DataStorageFormat::new(stable_file_version())
1015 } else {
1016 DataStorageFormat::new(ConcreteFileVersion::V1)
1017 }
1018 }
1019 }
1020 Some(format) => DataStorageFormat::try_from(format)?,
1021 };
1022
1023 let schema = Schema::try_from(fields_with_meta)?;
1024
1025 Ok(Self {
1026 schema,
1027 version: p.version,
1028 branch: p.branch,
1029 writer_version,
1030 version_aux_data: p.version_aux_data as usize,
1031 index_section: p.index_section.map(|i| i as usize),
1032 timestamp_nanos: timestamp_nanos.unwrap_or(0),
1033 tag: if p.tag.is_empty() { None } else { Some(p.tag) },
1034 reader_feature_flags: p.reader_feature_flags,
1035 writer_feature_flags: p.writer_feature_flags,
1036 max_fragment_id: p.max_fragment_id,
1037 fragments,
1038 transaction_file: if p.transaction_file.is_empty() {
1039 None
1040 } else {
1041 Some(p.transaction_file)
1042 },
1043 transaction_section: p.transaction_section.map(|i| i as usize),
1044 fragment_offsets,
1045 next_row_id: p.next_row_id,
1046 data_storage_format,
1047 config: p.config,
1048 table_metadata: p.table_metadata,
1049 base_paths: p
1050 .base_paths
1051 .iter()
1052 .map(|item| (item.id, item.clone().into()))
1053 .collect(),
1054 })
1055 }
1056}
1057
1058impl From<&Manifest> for pb::Manifest {
1059 fn from(m: &Manifest) -> Self {
1060 let timestamp_nanos = if m.timestamp_nanos == 0 {
1061 None
1062 } else {
1063 let nanos = m.timestamp_nanos % 1e9 as u128;
1064 let seconds = ((m.timestamp_nanos - nanos) / 1e9 as u128) as i64;
1065 Some(Timestamp {
1066 seconds,
1067 nanos: nanos as i32,
1068 })
1069 };
1070 let fields_with_meta: FieldsWithMeta = (&m.schema).into();
1071 Self {
1072 fields: fields_with_meta.fields.0,
1073 schema_metadata: m
1074 .schema
1075 .metadata
1076 .iter()
1077 .map(|(k, v)| (k.clone(), v.as_bytes().to_vec()))
1078 .collect(),
1079 version: m.version,
1080 branch: m.branch.clone(),
1081 writer_version: m
1082 .writer_version
1083 .as_ref()
1084 .map(|wv| pb::manifest::WriterVersion {
1085 library: wv.library.clone(),
1086 version: wv.version.clone(),
1087 prerelease: wv.prerelease.clone(),
1088 build_metadata: wv.build_metadata.clone(),
1089 }),
1090 fragments: m.fragments.iter().map(pb::DataFragment::from).collect(),
1091 fragment_tree: None,
1092 table_metadata: m.table_metadata.clone(),
1093 version_aux_data: m.version_aux_data as u64,
1094 index_section: m.index_section.map(|i| i as u64),
1095 timestamp: timestamp_nanos,
1096 tag: m.tag.clone().unwrap_or_default(),
1097 reader_feature_flags: m.reader_feature_flags,
1098 writer_feature_flags: m.writer_feature_flags,
1099 max_fragment_id: m.max_fragment_id,
1100 transaction_file: m.transaction_file.clone().unwrap_or_default(),
1101 next_row_id: m.next_row_id,
1102 data_format: Some(pb::manifest::DataStorageFormat {
1103 file_format: m.data_storage_format.file_format.clone(),
1104 version: m
1105 .data_storage_format
1106 .version
1107 .to_manifest_string()
1108 .to_string(),
1109 }),
1110 config: m.config.clone(),
1111 base_paths: m
1112 .base_paths
1113 .values()
1114 .map(|base_path| pb::BasePath {
1115 id: base_path.id,
1116 name: base_path.name.clone(),
1117 is_dataset_root: base_path.is_dataset_root,
1118 path: base_path.path.clone(),
1119 })
1120 .collect(),
1121 transaction_section: m.transaction_section.map(|i| i as u64),
1122 }
1123 }
1124}
1125
1126#[async_trait]
1127pub trait SelfDescribingFileReader {
1128 async fn try_new_self_described(
1136 object_store: &ObjectStore,
1137 path: &Path,
1138 cache: Option<&LanceCache>,
1139 ) -> Result<Self>
1140 where
1141 Self: Sized,
1142 {
1143 let reader = object_store.open(path).await?;
1144 Self::try_new_self_described_from_reader(reader.into(), cache).await
1145 }
1146
1147 async fn try_new_self_described_from_reader(
1148 reader: Arc<dyn Reader>,
1149 cache: Option<&LanceCache>,
1150 ) -> Result<Self>
1151 where
1152 Self: Sized;
1153}
1154
1155#[async_trait]
1156impl SelfDescribingFileReader for V1FileReader {
1157 async fn try_new_self_described_from_reader(
1158 reader: Arc<dyn Reader>,
1159 cache: Option<&LanceCache>,
1160 ) -> Result<Self> {
1161 let metadata = Self::read_metadata(reader.as_ref(), cache).await?;
1162 let manifest_position = metadata.manifest_position.ok_or(Error::internal(format!(
1163 "Attempt to open file at {} as self-describing but it did not contain a manifest",
1164 reader.path(),
1165 )))?;
1166 let mut manifest: Manifest = read_struct(reader.as_ref(), manifest_position).await?;
1167 populate_manifest_schema_dictionaries(&mut manifest, reader.as_ref()).await?;
1168 let schema = manifest.schema;
1169 let max_field_id = schema.max_field_id().unwrap_or_default();
1170 Self::try_new_from_reader(
1171 reader.path(),
1172 reader.clone(),
1173 Some(metadata),
1174 schema,
1175 0,
1176 0,
1177 max_field_id,
1178 cache,
1179 )
1180 .await
1181 }
1182}
1183
1184#[cfg(test)]
1185mod tests {
1186 use crate::feature_flags::FLAG_USE_V2_FORMAT_DEPRECATED;
1187 use crate::format::overlay::{DataOverlayFile, OverlayCoverage};
1188 use crate::format::{DataFile, DeletionFile, DeletionFileType};
1189 use std::num::NonZero;
1190
1191 use super::*;
1192
1193 use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
1194 use lance_core::datatypes::Field;
1195 use roaring::RoaringBitmap;
1196
1197 #[test]
1201 fn shallow_clone_stamps_base_id_on_overlay_files() {
1202 let arrow_schema = ArrowSchema::new(vec![ArrowField::new(
1203 "a",
1204 arrow_schema::DataType::Int64,
1205 false,
1206 )]);
1207 let schema = Schema::try_from(&arrow_schema).unwrap();
1208
1209 let mut fragment = Fragment::with_file_legacy(0, "base.lance", &schema, Some(10));
1210 fragment.overlays = vec![DataOverlayFile {
1211 data_file: DataFile::new_legacy_from_fields("overlay.lance", vec![0], None),
1212 coverage: OverlayCoverage::Shared(Arc::new(RoaringBitmap::from_iter([0_u32]))),
1213 committed_version: 1,
1214 }];
1215 let manifest = Manifest::new(
1216 schema,
1217 Arc::new(vec![fragment]),
1218 DataStorageFormat::default(),
1219 HashMap::new(),
1220 );
1221
1222 let cloned = manifest.shallow_clone(
1223 Some("parent".to_string()),
1224 "memory://parent".to_string(),
1225 7,
1226 None,
1227 String::new(),
1228 );
1229
1230 let fragment = &cloned.fragments[0];
1231 assert_eq!(fragment.files[0].base_id, Some(7));
1232 assert_eq!(
1233 fragment.overlays[0].data_file.base_id,
1234 Some(7),
1235 "the overlay's data file resolves against the parent as well"
1236 );
1237 }
1238
1239 #[test]
1240 fn old_empty_manifest_recovers_v1_or_current_stable() {
1241 let old_manifest = pb::Manifest {
1242 data_format: None,
1243 ..Default::default()
1244 };
1245 let recovered_v1 = Manifest::try_from(old_manifest.clone()).unwrap();
1246 assert_eq!(
1247 recovered_v1.data_storage_format.lance_file_format(),
1248 ConcreteFileVersion::V1
1249 );
1250
1251 let recovered_stable = Manifest::try_from(pb::Manifest {
1252 writer_feature_flags: FLAG_USE_V2_FORMAT_DEPRECATED,
1253 ..old_manifest
1254 })
1255 .unwrap();
1256 assert_eq!(
1257 recovered_stable.data_storage_format.lance_file_format(),
1258 stable_file_version()
1259 );
1260 }
1261
1262 #[test]
1263 fn manifest_persistence_rejects_selectors_and_public_aliases() {
1264 for version in ["stable", "next", "legacy", "0.3"] {
1265 let manifest = pb::Manifest {
1266 data_format: Some(pb::manifest::DataStorageFormat {
1267 file_format: LANCE_FORMAT_NAME.to_string(),
1268 version: version.to_string(),
1269 }),
1270 ..Default::default()
1271 };
1272 assert!(Manifest::try_from(manifest).is_err(), "accepted {version}");
1273 }
1274 }
1275
1276 #[test]
1277 fn manifest_codec_writes_canonical_exact_string() {
1278 let manifest = Manifest::new(
1279 Schema::default(),
1280 Arc::new(Vec::new()),
1281 DataStorageFormat::new(ConcreteFileVersion::V2_0),
1282 HashMap::new(),
1283 );
1284 let encoded = pb::Manifest::from(&manifest);
1285 assert_eq!(encoded.data_format.unwrap().version, "2.0");
1286 }
1287
1288 #[test]
1289 fn missing_format_infers_exact_version_and_rejects_mixed_files() {
1290 let v2_0 = Fragment::new(0).with_file(
1291 "v2_0.lance",
1292 vec![0],
1293 vec![0],
1294 ConcreteFileVersion::V2_0,
1295 None,
1296 );
1297 let manifest = Manifest::new(
1298 Schema::default(),
1299 Arc::new(vec![v2_0.clone()]),
1300 DataStorageFormat::new(ConcreteFileVersion::V1),
1301 HashMap::new(),
1302 );
1303 let mut encoded = pb::Manifest::from(&manifest);
1304 encoded.data_format = None;
1305 let recovered = Manifest::try_from(encoded).unwrap();
1306 assert_eq!(
1307 recovered.data_storage_format.lance_file_format(),
1308 ConcreteFileVersion::V2_0
1309 );
1310
1311 let v2_1 = Fragment::new(1).with_file(
1312 "v2_1.lance",
1313 vec![0],
1314 vec![0],
1315 ConcreteFileVersion::V2_1,
1316 None,
1317 );
1318 let mixed_manifest = Manifest::new(
1319 Schema::default(),
1320 Arc::new(vec![v2_0, v2_1]),
1321 DataStorageFormat::new(ConcreteFileVersion::V2_0),
1322 HashMap::new(),
1323 );
1324 let mut encoded = pb::Manifest::from(&mixed_manifest);
1325 encoded.data_format = None;
1326 let error = Manifest::try_from(encoded).unwrap_err();
1327 assert!(
1328 error
1329 .to_string()
1330 .contains("All data files must have the same version")
1331 );
1332 }
1333
1334 #[test]
1335 fn test_detach_sparse_inline_row_ids_copies_only_small_shares() {
1336 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("a", DataType::Int32, false)]);
1337 let buffer = Bytes::from(vec![7u8; 64]);
1339 let fragments = [0..10, 20..30].into_iter().enumerate().map(|(id, range)| {
1340 let mut fragment = Fragment::new(id as u64);
1341 fragment.row_id_meta = Some(RowIdMeta::Inline(InlineRowIds::from(buffer.slice(range))));
1342 fragment
1343 });
1344 let mut manifest = Manifest::new(
1345 Schema::try_from(&arrow_schema).unwrap(),
1346 Arc::new(fragments.collect()),
1347 DataStorageFormat::default(),
1348 HashMap::new(),
1349 );
1350 let ptr = |manifest: &Manifest, i: usize| match &manifest.fragments[i].row_id_meta {
1351 Some(RowIdMeta::Inline(data)) => data.bytes().as_ptr(),
1352 _ => unreachable!(),
1353 };
1354 let in_buffer = |p: *const u8| buffer.as_ptr_range().contains(&p);
1355
1356 manifest.detach_sparse_inline_row_ids(80);
1358 assert!(in_buffer(ptr(&manifest, 0)) && in_buffer(ptr(&manifest, 1)));
1359
1360 manifest.detach_sparse_inline_row_ids(81);
1362 assert!(!in_buffer(ptr(&manifest, 0)) && !in_buffer(ptr(&manifest, 1)));
1363 let Some(RowIdMeta::Inline(copied)) = &manifest.fragments[1].row_id_meta else {
1364 unreachable!()
1365 };
1366 assert_eq!(&**copied, &buffer[20..30]);
1367 }
1368
1369 #[test]
1370 fn test_writer_version() {
1371 let wv = WriterVersion::default();
1372 assert_eq!(wv.library, "lance");
1373
1374 let cargo_version = env!("CARGO_PKG_VERSION");
1376 let expected_tag = if cargo_version.contains('-') {
1377 Some(cargo_version.split('-').nth(1).unwrap())
1378 } else {
1379 None
1380 };
1381
1382 let version_parts: Vec<&str> = wv.version.split('.').collect();
1384 assert_eq!(
1385 version_parts.len(),
1386 3,
1387 "Version should be major.minor.patch"
1388 );
1389 assert!(
1390 !wv.version.contains('-'),
1391 "Version field should not contain prerelease"
1392 );
1393
1394 assert_eq!(wv.prerelease.as_deref(), expected_tag);
1396 assert_eq!(wv.build_metadata, None);
1398
1399 let version = wv.lance_lib_version().unwrap();
1401 assert_eq!(
1402 version.major,
1403 env!("CARGO_PKG_VERSION_MAJOR").parse::<u64>().unwrap()
1404 );
1405 assert_eq!(
1406 version.minor,
1407 env!("CARGO_PKG_VERSION_MINOR").parse::<u64>().unwrap()
1408 );
1409 assert_eq!(
1410 version.patch,
1411 env!("CARGO_PKG_VERSION_PATCH").parse::<u64>().unwrap() + 1
1413 );
1414 assert_eq!(version.pre.as_str(), expected_tag.unwrap_or(""));
1415
1416 for part in &[VersionPart::Major, VersionPart::Minor, VersionPart::Patch] {
1417 let mut bumped_version = version.clone();
1418 bump_version(&mut bumped_version, *part);
1419 assert!(version < bumped_version);
1420 }
1421 }
1422
1423 #[test]
1424 fn test_writer_version_split() {
1425 let (version, prerelease, build_metadata) =
1427 WriterVersion::split_version("2.0.0-rc.1").unwrap();
1428 assert_eq!(version, "2.0.0");
1429 assert_eq!(prerelease, Some("rc.1".to_string()));
1430 assert_eq!(build_metadata, None);
1431
1432 let (version, prerelease, build_metadata) = WriterVersion::split_version("2.0.0").unwrap();
1434 assert_eq!(version, "2.0.0");
1435 assert_eq!(prerelease, None);
1436 assert_eq!(build_metadata, None);
1437
1438 let (version, prerelease, build_metadata) =
1440 WriterVersion::split_version("2.0.0-rc.1+build.123").unwrap();
1441 assert_eq!(version, "2.0.0");
1442 assert_eq!(prerelease, Some("rc.1".to_string()));
1443 assert_eq!(build_metadata, Some("build.123".to_string()));
1444
1445 let (version, prerelease, build_metadata) =
1447 WriterVersion::split_version("2.0.0+build.123").unwrap();
1448 assert_eq!(version, "2.0.0");
1449 assert_eq!(prerelease, None);
1450 assert_eq!(build_metadata, Some("build.123".to_string()));
1451
1452 assert!(WriterVersion::split_version("not-a-version").is_none());
1454 }
1455
1456 #[test]
1457 fn test_writer_version_comparison_with_prerelease() {
1458 let v1 = WriterVersion {
1459 library: "lance".to_string(),
1460 version: "2.0.0".to_string(),
1461 prerelease: Some("rc.1".to_string()),
1462 build_metadata: None,
1463 };
1464
1465 let v2 = WriterVersion {
1466 library: "lance".to_string(),
1467 version: "2.0.0".to_string(),
1468 prerelease: None,
1469 build_metadata: None,
1470 };
1471
1472 let semver1 = v1.lance_lib_version().unwrap();
1473 let semver2 = v2.lance_lib_version().unwrap();
1474
1475 assert!(semver1 < semver2);
1477 }
1478
1479 #[test]
1480 fn test_writer_version_with_build_metadata() {
1481 let v = WriterVersion {
1482 library: "lance".to_string(),
1483 version: "2.0.0".to_string(),
1484 prerelease: Some("rc.1".to_string()),
1485 build_metadata: Some("build.123".to_string()),
1486 };
1487
1488 let semver = v.lance_lib_version().unwrap();
1489 assert_eq!(semver.to_string(), "2.0.0-rc.1+build.123");
1490 assert_eq!(semver.major, 2);
1491 assert_eq!(semver.minor, 0);
1492 assert_eq!(semver.patch, 0);
1493 assert_eq!(semver.pre.as_str(), "rc.1");
1494 assert_eq!(semver.build.as_str(), "build.123");
1495 }
1496
1497 #[test]
1498 fn test_writer_version_non_semver() {
1499 let v = WriterVersion {
1501 library: "lance".to_string(),
1502 version: "custom-build-v1".to_string(),
1503 prerelease: None,
1504 build_metadata: None,
1505 };
1506
1507 assert!(v.lance_lib_version().is_none());
1509
1510 assert_eq!(v.library, "lance");
1512 assert_eq!(v.version, "custom-build-v1");
1513 }
1514
1515 #[test]
1516 #[allow(deprecated)]
1517 fn test_older_than_with_prerelease() {
1518 let v_rc = WriterVersion {
1520 library: "lance".to_string(),
1521 version: "2.0.0".to_string(),
1522 prerelease: Some("rc.1".to_string()),
1523 build_metadata: None,
1524 };
1525
1526 assert!(v_rc.older_than(2, 0, 0));
1528
1529 assert!(v_rc.older_than(2, 0, 1));
1531
1532 assert!(!v_rc.older_than(1, 9, 9));
1534
1535 let v_release = WriterVersion {
1536 library: "lance".to_string(),
1537 version: "2.0.0".to_string(),
1538 prerelease: None,
1539 build_metadata: None,
1540 };
1541
1542 assert!(!v_release.older_than(2, 0, 0));
1544
1545 assert!(v_release.older_than(2, 0, 1));
1547 }
1548
1549 #[test]
1550 fn test_fragments_by_offset_range() {
1551 let arrow_schema = ArrowSchema::new(vec![ArrowField::new(
1552 "a",
1553 arrow_schema::DataType::Int64,
1554 false,
1555 )]);
1556 let schema = Schema::try_from(&arrow_schema).unwrap();
1557 let fragments = vec![
1558 Fragment::with_file_legacy(0, "path1", &schema, Some(10)),
1559 Fragment::with_file_legacy(1, "path2", &schema, Some(15)),
1560 Fragment::with_file_legacy(2, "path3", &schema, Some(20)),
1561 ];
1562 let manifest = Manifest::new(
1563 schema,
1564 Arc::new(fragments),
1565 DataStorageFormat::default(),
1566 HashMap::new(),
1567 );
1568
1569 let actual = manifest.fragments_by_offset_range(0..10);
1570 assert_eq!(actual.len(), 1);
1571 assert_eq!(actual[0].0, 0);
1572 assert_eq!(actual[0].1.id, 0);
1573
1574 let actual = manifest.fragments_by_offset_range(5..15);
1575 assert_eq!(actual.len(), 2);
1576 assert_eq!(actual[0].0, 0);
1577 assert_eq!(actual[0].1.id, 0);
1578 assert_eq!(actual[1].0, 10);
1579 assert_eq!(actual[1].1.id, 1);
1580
1581 let actual = manifest.fragments_by_offset_range(15..50);
1582 assert_eq!(actual.len(), 2);
1583 assert_eq!(actual[0].0, 10);
1584 assert_eq!(actual[0].1.id, 1);
1585 assert_eq!(actual[1].0, 25);
1586 assert_eq!(actual[1].1.id, 2);
1587
1588 let actual = manifest.fragments_by_offset_range(45..100);
1590 assert!(actual.is_empty());
1591
1592 assert!(manifest.fragments_by_offset_range(200..400).is_empty());
1593 }
1594
1595 #[test]
1596 fn test_max_field_id() {
1597 let mut field0 =
1599 Field::try_from(ArrowField::new("a", arrow_schema::DataType::Int64, false)).unwrap();
1600 field0.set_id(-1, &mut 0);
1601 let mut field2 =
1602 Field::try_from(ArrowField::new("b", arrow_schema::DataType::Int64, false)).unwrap();
1603 field2.set_id(-1, &mut 2);
1604
1605 let schema = Schema {
1606 fields: vec![field0, field2],
1607 metadata: Default::default(),
1608 };
1609 let fragments = vec![
1610 Fragment {
1611 id: 0,
1612 files: vec![DataFile::new_legacy_from_fields(
1613 "path1",
1614 vec![0, 1, 2],
1615 None,
1616 )],
1617 overlays: vec![],
1618 deletion_file: None,
1619 row_id_meta: None,
1620 physical_rows: None,
1621 created_at_version_meta: None,
1622 last_updated_at_version_meta: None,
1623 },
1624 Fragment {
1625 id: 1,
1626 files: vec![
1627 DataFile::new_legacy_from_fields("path2", vec![0, 1, 43], None),
1628 DataFile::new_legacy_from_fields("path3", vec![2], None),
1629 ],
1630 overlays: vec![],
1631 deletion_file: None,
1632 row_id_meta: None,
1633 physical_rows: None,
1634 created_at_version_meta: None,
1635 last_updated_at_version_meta: None,
1636 },
1637 ];
1638
1639 let manifest = Manifest::new(
1640 schema,
1641 Arc::new(fragments),
1642 DataStorageFormat::default(),
1643 HashMap::new(),
1644 );
1645
1646 assert_eq!(manifest.max_field_id(), 43);
1647 }
1648
1649 #[test]
1650 fn test_max_field_id_includes_overlay_files() {
1651 let mut field0 =
1652 Field::try_from(ArrowField::new("a", arrow_schema::DataType::Int64, false)).unwrap();
1653 field0.set_id(-1, &mut 0);
1654 let schema = Schema {
1655 fields: vec![field0],
1656 metadata: Default::default(),
1657 };
1658
1659 let mut fragment = Fragment {
1660 id: 0,
1661 files: vec![DataFile::new_legacy_from_fields("path1", vec![0], None)],
1662 overlays: vec![],
1663 deletion_file: None,
1664 row_id_meta: None,
1665 physical_rows: None,
1666 created_at_version_meta: None,
1667 last_updated_at_version_meta: None,
1668 };
1669 fragment.overlays = vec![DataOverlayFile {
1670 data_file: DataFile::new_legacy_from_fields("overlay.lance", vec![43], None),
1671 coverage: OverlayCoverage::Shared(Arc::new(RoaringBitmap::from_iter([0_u32]))),
1672 committed_version: 1,
1673 }];
1674
1675 let manifest = Manifest::new(
1676 schema,
1677 Arc::new(vec![fragment]),
1678 DataStorageFormat::default(),
1679 HashMap::new(),
1680 );
1681
1682 assert_eq!(manifest.max_field_id(), 43);
1683 }
1684
1685 #[test]
1686 fn test_config() {
1687 let arrow_schema = ArrowSchema::new(vec![ArrowField::new(
1688 "a",
1689 arrow_schema::DataType::Int64,
1690 false,
1691 )]);
1692 let schema = Schema::try_from(&arrow_schema).unwrap();
1693 let fragments = vec![
1694 Fragment::with_file_legacy(0, "path1", &schema, Some(10)),
1695 Fragment::with_file_legacy(1, "path2", &schema, Some(15)),
1696 Fragment::with_file_legacy(2, "path3", &schema, Some(20)),
1697 ];
1698 let mut manifest = Manifest::new(
1699 schema,
1700 Arc::new(fragments),
1701 DataStorageFormat::default(),
1702 HashMap::new(),
1703 );
1704
1705 let mut config = manifest.config.clone();
1706 config.insert("lance.test".to_string(), "value".to_string());
1707 config.insert("other-key".to_string(), "other-value".to_string());
1708
1709 manifest.config_mut().extend(config.clone());
1710 assert_eq!(manifest.config, config.clone());
1711
1712 config.remove("other-key");
1713 manifest.config_mut().remove("other-key");
1714 assert_eq!(manifest.config, config);
1715 }
1716
1717 #[test]
1718 fn test_manifest_summary() {
1719 let arrow_schema = ArrowSchema::new(vec![
1721 ArrowField::new("id", arrow_schema::DataType::Int64, false),
1722 ArrowField::new("name", arrow_schema::DataType::Utf8, true),
1723 ]);
1724 let schema = Schema::try_from(&arrow_schema).unwrap();
1725
1726 let empty_manifest = Manifest::new(
1727 schema.clone(),
1728 Arc::new(vec![]),
1729 DataStorageFormat::default(),
1730 HashMap::new(),
1731 );
1732
1733 let empty_summary = empty_manifest.summary();
1734 assert_eq!(empty_summary.total_rows, 0);
1735 assert_eq!(empty_summary.total_files_size, 0);
1736 assert_eq!(empty_summary.total_fragments, 0);
1737 assert_eq!(empty_summary.total_data_files, 0);
1738 assert_eq!(empty_summary.total_deletion_file_rows, 0);
1739 assert_eq!(empty_summary.total_data_file_rows, 0);
1740 assert_eq!(empty_summary.total_deletion_files, 0);
1741
1742 let empty_fragments = vec![
1744 Fragment::with_file_legacy(0, "empty_file1.lance", &schema, Some(0)),
1745 Fragment::with_file_legacy(1, "empty_file2.lance", &schema, Some(0)),
1746 ];
1747
1748 let empty_files_manifest = Manifest::new(
1749 schema.clone(),
1750 Arc::new(empty_fragments),
1751 DataStorageFormat::default(),
1752 HashMap::new(),
1753 );
1754
1755 let empty_files_summary = empty_files_manifest.summary();
1756 assert_eq!(empty_files_summary.total_rows, 0);
1757 assert_eq!(empty_files_summary.total_files_size, 0);
1758 assert_eq!(empty_files_summary.total_fragments, 2);
1759 assert_eq!(empty_files_summary.total_data_files, 2);
1760 assert_eq!(empty_files_summary.total_deletion_file_rows, 0);
1761 assert_eq!(empty_files_summary.total_data_file_rows, 0);
1762 assert_eq!(empty_files_summary.total_deletion_files, 0);
1763
1764 let real_fragments = vec![
1766 Fragment::with_file_legacy(0, "data_file1.lance", &schema, Some(100)),
1767 Fragment::with_file_legacy(1, "data_file2.lance", &schema, Some(250)),
1768 Fragment::with_file_legacy(2, "data_file3.lance", &schema, Some(75)),
1769 ];
1770
1771 let real_data_manifest = Manifest::new(
1772 schema.clone(),
1773 Arc::new(real_fragments),
1774 DataStorageFormat::default(),
1775 HashMap::new(),
1776 );
1777
1778 let real_data_summary = real_data_manifest.summary();
1779 assert_eq!(real_data_summary.total_rows, 425); assert_eq!(real_data_summary.total_files_size, 0); assert_eq!(real_data_summary.total_fragments, 3);
1782 assert_eq!(real_data_summary.total_data_files, 3);
1783 assert_eq!(real_data_summary.total_deletion_file_rows, 0);
1784 assert_eq!(real_data_summary.total_data_file_rows, 425);
1785 assert_eq!(real_data_summary.total_deletion_files, 0);
1786
1787 let mut fragment_with_deletion = Fragment::new(0)
1789 .with_file(
1790 "data_with_deletion.lance",
1791 vec![0, 1],
1792 vec![0, 1],
1793 stable_file_version(),
1794 NonZero::new(1000),
1795 )
1796 .with_physical_rows(50);
1797 fragment_with_deletion.deletion_file = Some(DeletionFile {
1798 read_version: 123,
1799 id: 456,
1800 file_type: DeletionFileType::Array,
1801 num_deleted_rows: Some(10),
1802 base_id: None,
1803 });
1804
1805 let manifest_with_deletion = Manifest::new(
1806 schema,
1807 Arc::new(vec![fragment_with_deletion]),
1808 DataStorageFormat::default(),
1809 HashMap::new(),
1810 );
1811
1812 let deletion_summary = manifest_with_deletion.summary();
1813 assert_eq!(deletion_summary.total_rows, 40); assert_eq!(deletion_summary.total_files_size, 1000);
1815 assert_eq!(deletion_summary.total_fragments, 1);
1816 assert_eq!(deletion_summary.total_data_files, 1);
1817 assert_eq!(deletion_summary.total_deletion_file_rows, 10);
1818 assert_eq!(deletion_summary.total_data_file_rows, 50);
1819 assert_eq!(deletion_summary.total_deletion_files, 1);
1820
1821 let stats_map: BTreeMap<String, String> = deletion_summary.into();
1823 assert_eq!(stats_map.len(), 7)
1824 }
1825}