molframe_core/provider/
catalog.rs

1//! Compact dataset metadata without per-chunk materialisation.
2
3use super::{ChunkDescriptor, ChunkId, DatasetId, LogicalRow, ProviderError};
4use std::sync::Arc;
5
6/// Payload carried by a dataset.
7#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
8pub enum PayloadKind {
9    /// Atom topology and base coordinates.
10    Structure,
11    /// One typed property column.
12    Property,
13    /// Coordinates for one frame.
14    Frame,
15    /// Chemical bond rows with globally addressable atom endpoints.
16    BondTopology,
17}
18
19/// How chunk boundaries are determined.
20#[derive(Clone, Copy, Debug, PartialEq, Eq)]
21pub enum ChunkLayout {
22    /// Every chunk except the tail has the same row count.
23    Regular {
24        /// Rows in a full chunk.
25        rows_per_chunk: u32,
26    },
27    /// The source provides exact boundaries, while this value guides demand.
28    SourceDefined {
29        /// Preferred maximum rows requested from the source.
30        target_rows: u32,
31    },
32}
33
34impl ChunkLayout {
35    /// Preferred row count used for bounded pull demand.
36    #[must_use]
37    pub const fn target_rows(self) -> u32 {
38        match self {
39            Self::Regular { rows_per_chunk } => rows_per_chunk,
40            Self::SourceDefined { target_rows } => target_rows,
41        }
42    }
43}
44
45/// O(1) metadata for one logical dataset.
46#[derive(Clone, Copy, Debug, PartialEq, Eq)]
47pub struct DatasetDescriptor {
48    id: DatasetId,
49    payload: PayloadKind,
50    logical_rows: u64,
51    chunk_count: u64,
52    first_chunk: ChunkId,
53    layout: ChunkLayout,
54}
55
56impl DatasetDescriptor {
57    /// Describes a regularly chunked logical dataset.
58    ///
59    /// # Errors
60    ///
61    /// Returns a typed error for zero chunk size or 64-bit arithmetic overflow.
62    pub fn regular(
63        id: DatasetId,
64        payload: PayloadKind,
65        logical_rows: u64,
66        rows_per_chunk: u32,
67        first_chunk: ChunkId,
68    ) -> Result<Self, ProviderError> {
69        if rows_per_chunk == 0 {
70            return Err(ProviderError::ZeroChunkRows { dataset: id });
71        }
72        let divisor = u64::from(rows_per_chunk);
73        let chunk_count = match logical_rows.checked_add(divisor - 1) {
74            Some(rounded) => rounded / divisor,
75            None => {
76                return Err(ProviderError::IdentityOverflow {
77                    dataset: id,
78                    field: "chunk count",
79                });
80            }
81        };
82        Self::new(
83            id,
84            payload,
85            logical_rows,
86            chunk_count,
87            first_chunk,
88            ChunkLayout::Regular { rows_per_chunk },
89        )
90    }
91
92    /// Describes source-defined chunk boundaries without storing each boundary.
93    ///
94    /// # Errors
95    ///
96    /// Returns a typed error for contradictory counts, zero target size, or
97    /// chunk-identity overflow.
98    pub fn source_defined(
99        id: DatasetId,
100        payload: PayloadKind,
101        logical_rows: u64,
102        chunk_count: u64,
103        first_chunk: ChunkId,
104        target_rows: u32,
105    ) -> Result<Self, ProviderError> {
106        Self::new(
107            id,
108            payload,
109            logical_rows,
110            chunk_count,
111            first_chunk,
112            ChunkLayout::SourceDefined { target_rows },
113        )
114    }
115
116    fn new(
117        id: DatasetId,
118        payload: PayloadKind,
119        logical_rows: u64,
120        chunk_count: u64,
121        first_chunk: ChunkId,
122        layout: ChunkLayout,
123    ) -> Result<Self, ProviderError> {
124        if layout.target_rows() == 0 {
125            return Err(ProviderError::ZeroChunkRows { dataset: id });
126        }
127        if logical_rows > 0 && chunk_count == 0 {
128            return Err(ProviderError::MissingChunks {
129                dataset: id,
130                rows: logical_rows,
131            });
132        }
133        validate_chunk_range(id, first_chunk, chunk_count)?;
134        Ok(Self {
135            id,
136            payload,
137            logical_rows,
138            chunk_count,
139            first_chunk,
140            layout,
141        })
142    }
143
144    /// Stable dataset identity.
145    #[must_use]
146    pub const fn id(self) -> DatasetId {
147        self.id
148    }
149
150    /// Typed payload produced for this dataset.
151    #[must_use]
152    pub const fn payload(self) -> PayloadKind {
153        self.payload
154    }
155
156    /// Total logical rows, which may exceed `u32::MAX`.
157    #[must_use]
158    pub const fn logical_rows(self) -> u64 {
159        self.logical_rows
160    }
161
162    /// Number of chunks, represented without a per-chunk vector.
163    #[must_use]
164    pub const fn chunk_count(self) -> u64 {
165        self.chunk_count
166    }
167
168    /// First stable chunk identity.
169    #[must_use]
170    pub const fn first_chunk(self) -> ChunkId {
171        self.first_chunk
172    }
173
174    /// Chunk-boundary policy.
175    #[must_use]
176    pub const fn layout(self) -> ChunkLayout {
177        self.layout
178    }
179
180    /// Resolves a regular chunk in constant time.
181    ///
182    /// # Errors
183    ///
184    /// Returns [`ProviderError::UnknownChunk`] for a foreign chunk or when the
185    /// source owns its own boundaries.
186    pub fn regular_chunk(self, chunk: ChunkId) -> Result<ChunkDescriptor, ProviderError> {
187        let ChunkLayout::Regular { rows_per_chunk } = self.layout else {
188            return Err(ProviderError::UnknownChunk {
189                dataset: self.id,
190                chunk,
191            });
192        };
193        let ordinal = self.chunk_ordinal(chunk)?;
194        let start = ordinal.checked_mul(u64::from(rows_per_chunk)).ok_or(
195            ProviderError::IdentityOverflow {
196                dataset: self.id,
197                field: "logical row",
198            },
199        )?;
200        let remaining =
201            self.logical_rows
202                .checked_sub(start)
203                .ok_or(ProviderError::LogicalRowOutOfRange {
204                    dataset: self.id,
205                    row: LogicalRow::new(start),
206                })?;
207        let rows = remaining.min(u64::from(rows_per_chunk));
208        let rows = u32::try_from(rows).map_err(|_| ProviderError::IdentityOverflow {
209            dataset: self.id,
210            field: "local row count",
211        })?;
212        ChunkDescriptor::new(self.id, chunk, LogicalRow::new(start), rows)
213    }
214
215    pub(crate) fn chunk_ordinal(self, chunk: ChunkId) -> Result<u64, ProviderError> {
216        let Some(ordinal) = chunk.get().checked_sub(self.first_chunk.get()) else {
217            return Err(ProviderError::UnknownChunk {
218                dataset: self.id,
219                chunk,
220            });
221        };
222        if ordinal >= self.chunk_count {
223            return Err(ProviderError::UnknownChunk {
224                dataset: self.id,
225                chunk,
226            });
227        }
228        Ok(ordinal)
229    }
230}
231
232/// Hierarchical catalog whose memory scales with datasets, not logical chunks.
233#[derive(Clone, Debug, Default)]
234pub struct DatasetCatalog {
235    datasets: Arc<[DatasetDescriptor]>,
236}
237
238impl DatasetCatalog {
239    /// Validates, sorts, and shares dataset metadata.
240    ///
241    /// # Errors
242    ///
243    /// Rejects duplicate dataset identities and overlapping chunk namespaces.
244    pub fn new(mut datasets: Vec<DatasetDescriptor>) -> Result<Self, ProviderError> {
245        datasets.sort_unstable_by_key(|dataset| dataset.id());
246        validate_dataset_ids(&datasets)?;
247        validate_chunk_ranges(&datasets)?;
248        Ok(Self {
249            datasets: datasets.into(),
250        })
251    }
252
253    /// Number of logical datasets.
254    #[must_use]
255    pub fn len(&self) -> usize {
256        self.datasets.len()
257    }
258
259    /// Reports whether the catalog contains no datasets.
260    #[must_use]
261    pub fn is_empty(&self) -> bool {
262        self.datasets.is_empty()
263    }
264
265    /// Shared sorted descriptors.
266    #[must_use]
267    pub fn datasets(&self) -> &[DatasetDescriptor] {
268        &self.datasets
269    }
270
271    /// Finds one dataset in logarithmic time.
272    #[must_use]
273    pub fn get(&self, id: DatasetId) -> Option<DatasetDescriptor> {
274        self.datasets
275            .binary_search_by_key(&id, |dataset| dataset.id())
276            .ok()
277            .and_then(|index| self.datasets.get(index).copied())
278    }
279}
280
281fn validate_chunk_range(
282    dataset: DatasetId,
283    first: ChunkId,
284    count: u64,
285) -> Result<(), ProviderError> {
286    if count == 0 {
287        return Ok(());
288    }
289    first
290        .get()
291        .checked_add(count - 1)
292        .map(|_| ())
293        .ok_or(ProviderError::IdentityOverflow {
294            dataset,
295            field: "chunk identity",
296        })
297}
298
299fn validate_dataset_ids(datasets: &[DatasetDescriptor]) -> Result<(), ProviderError> {
300    for pair in datasets.windows(2) {
301        if pair[0].id() == pair[1].id() {
302            return Err(ProviderError::DuplicateDataset {
303                dataset: pair[0].id(),
304            });
305        }
306    }
307    Ok(())
308}
309
310fn validate_chunk_ranges(datasets: &[DatasetDescriptor]) -> Result<(), ProviderError> {
311    let mut ranges: Vec<_> = datasets
312        .iter()
313        .filter(|dataset| dataset.chunk_count() > 0)
314        .map(|dataset| {
315            let end = dataset.first_chunk().get() + dataset.chunk_count() - 1;
316            (dataset.first_chunk().get(), end)
317        })
318        .collect();
319    ranges.sort_unstable_by_key(|range| range.0);
320    for pair in ranges.windows(2) {
321        if pair[1].0 <= pair[0].1 {
322            return Err(ProviderError::OverlappingChunkRange {
323                chunk: ChunkId::new(pair[1].0),
324            });
325        }
326    }
327    Ok(())
328}
329
330#[cfg(test)]
331#[path = "catalog_tests.rs"]
332mod tests;
molframe_core/provider/catalog.rs

molframe_core/provider/
catalog.rs
