fcb_core/reader/
mod.rs

1pub mod city_buffer;
2pub mod deserializer;
3use crate::static_btree::Offset;
4use city_buffer::*;
5use cjseq::CityJSONFeature;
6use deserializer::to_cj_feature;
7
8use crate::error::Error;
9use crate::fb::{size_prefixed_root_as_city_feature, CityFeature};
10use crate::packed_rtree::{self, PackedRTree, Query};
11use crate::{check_magic_bytes, size_prefixed_root_as_header, Header, HEADER_MAX_BUFFER_SIZE};
12use fallible_streaming_iterator::FallibleStreamingIterator;
13use std::io::{self, Read, Seek, SeekFrom, Write};
14mod attr_query;
15pub mod geom_decoder;
16pub use attr_query::*;
17use std::marker::PhantomData;
18mod meta;
19pub use meta::{Column as MetaColumn, ColumnType as MetaColumnType, Meta};
20
21pub struct FcbReader<R> {
22    reader: R,
23    verify: bool,
24    buffer: FcbBuffer,
25}
26
27pub struct FeatureIter<R, S> {
28    reader: R,
29    /// FlatBuffers verification
30    verify: bool,
31    // feature reading requires header access, therefore
32    // header_buf is included in the FgbFeature struct.
33    buffer: FcbBuffer,
34    /// Select>ed features or None if no bbox filter
35    item_filter: Option<Vec<packed_rtree::SearchResultItem>>,
36    /// Selected attributes or None if no attribute filter
37    item_attr_filter: Option<Vec<Offset>>,
38    /// Number of selected features (None for undefined feature count)
39    count: Option<usize>,
40    /// Current feature number
41    feat_no: usize,
42    /// File offset within feature section
43    cur_pos: u64,
44    /// Reading state
45    state: State,
46    /// Whether or not the underlying reader is Seek
47    seekable_marker: PhantomData<S>,
48    feature_offset: FeatureOffset,
49    total_feat_count: u64,
50}
51
52#[doc(hidden)]
53pub(super) struct FeatureOffset {
54    magic_bytes: u64,
55    header: u64,
56    rtree_index: u64,
57    attributes: u64,
58}
59
60#[derive(Debug, PartialEq, Eq)]
61enum State {
62    Init,
63    ReadFirstFeatureSize,
64    Reading,
65    Finished,
66}
67
68#[doc(hidden)]
69pub mod reader_trait {
70    pub struct Seekable;
71    pub struct NotSeekable;
72}
73use reader_trait::*;
74
75impl<R: Read> FcbReader<R> {
76    pub fn open(reader: R) -> Result<FcbReader<R>, Error> {
77        let reader = Self::read_header(reader, true)?;
78        Ok(reader)
79    }
80
81    /// Open a reader without verifying the FlatBuffers data.
82    ///
83    /// # Safety
84    /// This function skips FlatBuffers verification. The caller must ensure that the input data
85    /// is valid and properly formatted to avoid undefined behavior.
86    pub unsafe fn open_unchecked(reader: R) -> Result<FcbReader<R>, Error> {
87        Self::read_header(reader, false)
88    }
89
90    fn read_header(mut reader: R, verify: bool) -> Result<FcbReader<R>, Error> {
91        let mut magic_buf: [u8; 8] = [0; 8];
92        reader.read_exact(&mut magic_buf)?;
93        if !check_magic_bytes(&magic_buf) {
94            return Err(Error::MissingMagicBytes);
95        }
96
97        let mut size_buf: [u8; 4] = [0; 4]; // MEMO: 4 bytes for size prefix. This is comvention for FlatBuffers's size_prefixed_root
98        reader.read_exact(&mut size_buf)?;
99        let header_size = u32::from_le_bytes(size_buf) as usize;
100        if !((8..=HEADER_MAX_BUFFER_SIZE).contains(&header_size)) {
101            return Err(Error::IllegalHeaderSize(header_size));
102        }
103
104        let mut header_buf = Vec::with_capacity(header_size + 4); // 4 bytes for size prefix
105        header_buf.extend_from_slice(&size_buf);
106        header_buf.resize(header_buf.capacity(), 0);
107        reader.read_exact(&mut header_buf[4..])?;
108
109        if verify {
110            let _header = size_prefixed_root_as_header(&header_buf);
111        }
112
113        Ok(FcbReader {
114            reader,
115            verify,
116            buffer: FcbBuffer {
117                header_buf,
118                features_buf: Vec::new(),
119            },
120        })
121    }
122
123    pub fn select_all_seq(mut self) -> Result<FeatureIter<R, NotSeekable>, Error> {
124        let index_size = self.attr_index_size() + self.rtree_index_size();
125        // discard bufer of index
126        io::copy(&mut (&mut self.reader).take(index_size), &mut io::sink())?;
127        let feature_offset = FeatureOffset {
128            magic_bytes: 8,
129            header: 4 + self.buffer.header_buf.len() as u64,
130            rtree_index: self.rtree_index_size(),
131            attributes: self.attr_index_size(),
132        };
133        let total_feat_count = self.buffer.header().features_count();
134        Ok(FeatureIter::new(
135            self.reader,
136            self.verify,
137            self.buffer,
138            None,
139            None,
140            feature_offset,
141            total_feat_count,
142        ))
143    }
144
145    pub fn select_query_seq(mut self, query: Query) -> Result<FeatureIter<R, NotSeekable>, Error> {
146        // Read R-Tree index and build filter for features within bbox
147        let header = self.buffer.header();
148        if header.index_node_size() == 0 || header.features_count() == 0 {
149            return Err(Error::NoIndex);
150        }
151        let index = PackedRTree::from_buf(
152            &mut self.reader,
153            header.features_count() as usize,
154            header.index_node_size(),
155        )?;
156        let list = index.search(query)?;
157        debug_assert!(
158            list.windows(2).all(|w| w[0].offset < w[1].offset),
159            "Since the tree is traversed breadth first, list should be sorted by construction."
160        );
161        // skip attribute index
162        let index_size = self.attr_index_size();
163        io::copy(&mut (&mut self.reader).take(index_size), &mut io::sink())?;
164        let feature_offset = FeatureOffset {
165            magic_bytes: 8,
166            header: 4 + self.buffer.header_buf.len() as u64,
167            rtree_index: self.rtree_index_size(),
168            attributes: self.attr_index_size(),
169        };
170        let total_feat_count = list.len() as u64;
171        Ok(FeatureIter::new(
172            self.reader,
173            self.verify,
174            self.buffer,
175            Some(list),
176            None,
177            feature_offset,
178            total_feat_count,
179        ))
180    }
181}
182
183impl<R: Read + Seek> FcbReader<R> {
184    pub fn select_all(mut self) -> Result<FeatureIter<R, Seekable>, Error> {
185        // skip index
186        let feature_offset = FeatureOffset {
187            magic_bytes: 8,
188            header: 4 + self.buffer.header_buf.len() as u64,
189            rtree_index: self.rtree_index_size(),
190            attributes: self.attr_index_size(),
191        };
192        let index_size = self.attr_index_size() + self.rtree_index_size();
193        self.reader.seek(SeekFrom::Current(index_size as i64))?;
194        let total_feat_count = self.buffer.header().features_count();
195        Ok(FeatureIter::new(
196            self.reader,
197            self.verify,
198            self.buffer,
199            None,
200            None,
201            feature_offset,
202            total_feat_count,
203        ))
204    }
205
206    pub fn select_query(
207        mut self,
208        query: Query,
209        limit: Option<usize>,
210        offset: Option<usize>,
211    ) -> Result<FeatureIter<R, Seekable>, Error> {
212        // Read R-Tree index and build filter for features within bbox
213        let header = self.buffer.header();
214        if header.index_node_size() == 0 || header.features_count() == 0 {
215            return Err(Error::NoIndex);
216        }
217        // The R-tree branching factor is a per-file property: traversing with
218        // the compile-time default instead walks the wrong node ranges.
219        let list = PackedRTree::stream_search(
220            &mut self.reader,
221            header.features_count() as usize,
222            header.index_node_size(),
223            query,
224        )?;
225        let list: Vec<_> = list
226            .into_iter()
227            .skip(offset.unwrap_or(0))
228            .take(limit.unwrap_or(usize::MAX))
229            .collect();
230        debug_assert!(
231            list.windows(2).all(|w| w[0].offset < w[1].offset),
232            "Since the tree is traversed breadth first, list should be sorted by construction."
233        );
234
235        // skip index
236        self.reader
237            .seek(SeekFrom::Current(self.attr_index_size() as i64))?;
238        let feature_offset = FeatureOffset {
239            magic_bytes: 8,
240            header: 4 + self.buffer.header_buf.len() as u64,
241            rtree_index: self.rtree_index_size(),
242            attributes: self.attr_index_size(),
243        };
244        let total_feat_count = list.len() as u64;
245        Ok(FeatureIter::new(
246            self.reader,
247            self.verify,
248            self.buffer,
249            Some(list),
250            None,
251            feature_offset,
252            total_feat_count,
253        ))
254    }
255}
256
257impl<R: Read> FcbReader<R> {
258    pub fn header(&self) -> Header<'_> {
259        self.buffer.header()
260    }
261
262    pub fn root_attr_schema(
263        &self,
264    ) -> Option<flatbuffers::Vector<'_, flatbuffers::ForwardsUOffset<crate::fb::Column<'_>>>> {
265        self.buffer.header().columns()
266    }
267
268    fn rtree_index_size(&self) -> u64 {
269        let header = self.buffer.header();
270        let feat_count = header.features_count() as usize;
271        if header.index_node_size() > 0 && feat_count > 0 {
272            PackedRTree::index_size(feat_count, header.index_node_size()) as u64
273        } else {
274            0
275        }
276    }
277
278    fn attr_index_size(&self) -> u64 {
279        let header = self.buffer.header();
280        let len = header
281            .attribute_index()
282            .map(|attr_index| {
283                attr_index
284                    .iter()
285                    .try_fold(0u64, |acc, ai| {
286                        let len = ai.length() as u64;
287                        if len > u64::MAX - acc {
288                            Err(Error::AttributeIndexSizeOverflow)
289                        } else {
290                            Ok(acc + len)
291                        }
292                    })
293                    .unwrap_or(0)
294            })
295            .unwrap_or(0);
296        len
297    }
298}
299
300impl FeatureOffset {
301    fn total_size(&self) -> u64 {
302        self.magic_bytes + self.header + self.rtree_index + self.attributes
303    }
304}
305
306impl<R: Read> FallibleStreamingIterator for FeatureIter<R, NotSeekable> {
307    type Item = FcbBuffer;
308    type Error = Error;
309
310    fn advance(&mut self) -> Result<(), Error> {
311        if self.advance_finished() {
312            return Ok(());
313        }
314        if let Some(filter) = &self.item_filter {
315            let item = &filter[self.feat_no];
316            if item.offset as u64 > self.cur_pos {
317                if self.state == State::ReadFirstFeatureSize {
318                    self.state = State::Reading;
319                }
320                // skip features
321                let seek_bytes = item.offset as u64 - self.cur_pos;
322                io::copy(&mut (&mut self.reader).take(seek_bytes), &mut io::sink())?;
323                self.cur_pos += seek_bytes;
324            }
325        }
326
327        if let Some(attr_filter) = &self.item_attr_filter {
328            let item_offset = attr_filter[self.feat_no];
329            // only skip if we haven't reached the attribute offset yet
330            if item_offset > self.cur_pos {
331                if self.state == State::ReadFirstFeatureSize {
332                    self.state = State::Reading;
333                }
334                let seek_bytes = item_offset - self.cur_pos;
335                io::copy(&mut (&mut self.reader).take(seek_bytes), &mut io::sink())?;
336                self.cur_pos += seek_bytes;
337            }
338        }
339
340        self.read_feature()
341    }
342
343    fn get(&self) -> Option<&FcbBuffer> {
344        self.iter_get()
345    }
346
347    fn size_hint(&self) -> (usize, Option<usize>) {
348        self.iter_size_hint()
349    }
350}
351
352impl<R: Read + Seek> FallibleStreamingIterator for FeatureIter<R, Seekable> {
353    type Item = FcbBuffer;
354    type Error = Error;
355
356    fn advance(&mut self) -> Result<(), Error> {
357        if self.advance_finished() {
358            return Ok(());
359        }
360        if let Some(filter) = &self.item_filter {
361            let item = &filter[self.feat_no];
362            if item.offset as u64 > self.cur_pos {
363                if self.state == State::ReadFirstFeatureSize {
364                    self.state = State::Reading;
365                }
366                // skip features
367                let seek_bytes = item.offset as u64 - self.cur_pos;
368                self.reader.seek(SeekFrom::Current(seek_bytes as i64))?;
369                self.cur_pos += seek_bytes;
370            }
371        }
372
373        if let Some(attr_filter) = &self.item_attr_filter {
374            if self.state == State::ReadFirstFeatureSize {
375                self.state = State::Reading;
376            }
377            let item_offset = attr_filter[self.feat_no];
378            self.reader.seek(SeekFrom::Start(
379                self.feature_offset.total_size() + item_offset,
380            ))?;
381            self.cur_pos = item_offset;
382        }
383
384        self.read_feature()
385    }
386
387    fn get(&self) -> Option<&FcbBuffer> {
388        self.iter_get()
389    }
390
391    fn size_hint(&self) -> (usize, Option<usize>) {
392        self.iter_size_hint()
393    }
394}
395
396impl<R: Read> FeatureIter<R, NotSeekable> {
397    pub fn cur_feature(&self) -> CityFeature<'_> {
398        self.buffer.feature()
399    }
400
401    pub fn cur_cj_feature(&self) -> Result<CityJSONFeature, Error> {
402        let fcb_feature = self.buffer.feature();
403        let root_attr_schema = self.buffer.header().columns();
404        let semantic_attr_schema = self.buffer.header().semantic_columns();
405
406        to_cj_feature(fcb_feature, root_attr_schema, semantic_attr_schema)
407    }
408
409    pub fn get_features(&mut self) -> Result<Vec<CityFeature<'_>>, Error> {
410        // Ok(features)
411        todo!("implement")
412    }
413
414    #[allow(clippy::should_implement_trait)]
415    pub fn next(&mut self) -> Result<Option<&Self>, Error> {
416        self.advance()?;
417        if self.get().is_some() {
418            Ok(Some(self))
419        } else {
420            Ok(None)
421        }
422    }
423}
424
425impl<R: Read + Seek> FeatureIter<R, Seekable> {
426    pub fn cur_feature(&self) -> CityFeature<'_> {
427        self.buffer.feature()
428    }
429    pub fn cur_feature_len(&self) -> usize {
430        self.buffer.features_buf.len()
431    }
432    /// Return current feature
433    pub fn cur_cj_feature(&self) -> Result<CityJSONFeature, Error> {
434        let fcb_feature = self.buffer.feature();
435        let root_attr_schema = self.buffer.header().columns();
436        let semantic_attr_schema = self.buffer.header().semantic_columns();
437        to_cj_feature(fcb_feature, root_attr_schema, semantic_attr_schema)
438    }
439
440    pub fn get_features(&mut self, _: impl Write) -> Result<(), Error> {
441        todo!("implement")
442    }
443
444    pub fn get_current_feature(&self) -> CityFeature<'_> {
445        self.buffer.feature()
446    }
447
448    #[allow(clippy::should_implement_trait)]
449    pub fn next(&mut self) -> Result<Option<&Self>, Error> {
450        self.advance()?;
451        if self.get().is_some() {
452            Ok(Some(self))
453        } else {
454            Ok(None)
455        }
456    }
457}
458
459impl<R: Read, S> FeatureIter<R, S> {
460    pub(super) fn new(
461        reader: R,
462        verify: bool,
463        buffer: FcbBuffer,
464        item_filter: Option<Vec<packed_rtree::SearchResultItem>>,
465        item_attr_filter: Option<Vec<Offset>>,
466        feature_offset: FeatureOffset,
467        total_feat_count: u64,
468    ) -> FeatureIter<R, S> {
469        let mut iter = FeatureIter {
470            reader,
471            verify,
472            buffer,
473            item_filter,
474            item_attr_filter,
475            count: None,
476            feat_no: 0,
477            cur_pos: 0,
478            state: State::Init,
479            seekable_marker: PhantomData,
480            feature_offset,
481            total_feat_count,
482        };
483
484        if iter.read_feature_size() {
485            iter.state = State::Finished;
486        } else {
487            iter.state = State::ReadFirstFeatureSize
488        }
489
490        iter.count = match &iter.item_filter {
491            Some(list) => Some(list.len()),
492            None => {
493                let feat_count = iter.buffer.header().features_count() as usize;
494                if feat_count > 0 {
495                    Some(feat_count)
496                } else if iter.state == State::Finished {
497                    Some(0)
498                } else {
499                    None
500                }
501            }
502        };
503
504        iter
505    }
506
507    pub fn header(&self) -> Header<'_> {
508        self.buffer.header()
509    }
510
511    pub fn root_attr_schema(
512        &self,
513    ) -> Option<flatbuffers::Vector<'_, flatbuffers::ForwardsUOffset<crate::fb::Column<'_>>>> {
514        self.buffer.header().columns()
515    }
516
517    pub fn features_count(&self) -> Option<usize> {
518        Some(self.total_feat_count as usize)
519    }
520
521    fn advance_finished(&mut self) -> bool {
522        if self.state == State::Finished {
523            return true;
524        }
525        if let Some(count) = self.count {
526            if self.feat_no >= count {
527                self.state = State::Finished;
528                return true;
529            }
530        }
531        if let Some(attr_filter) = &self.item_attr_filter {
532            if self.feat_no >= attr_filter.len() {
533                self.state = State::Finished;
534                return true;
535            }
536        }
537        false
538    }
539
540    /// Read feature size and return true if end of dataset reached
541    fn read_feature_size(&mut self) -> bool {
542        self.buffer.features_buf.resize(4, 0);
543        self.cur_pos += 4;
544        self.reader
545            .read_exact(&mut self.buffer.features_buf)
546            .is_err()
547    }
548
549    fn read_feature(&mut self) -> Result<(), Error> {
550        match self.state {
551            State::ReadFirstFeatureSize => {
552                self.state = State::Reading;
553            }
554            State::Reading => {
555                if self.read_feature_size() {
556                    self.state = State::Finished;
557                    return Ok(());
558                }
559            }
560            State::Finished => {
561                debug_assert!(
562                    false,
563                    "shouldn't call read_feature on already finished Iter"
564                );
565                return Ok(());
566            }
567            State::Init => {
568                unreachable!("should have read first feature size before reading any features")
569            }
570        }
571        let sbuf = &self.buffer.features_buf;
572        let feature_size = u32::from_le_bytes([sbuf[0], sbuf[1], sbuf[2], sbuf[3]]) as usize;
573        self.buffer.features_buf.resize(feature_size + 4, 0);
574        self.reader.read_exact(&mut self.buffer.features_buf[4..])?;
575        if self.verify {
576            let _feature = size_prefixed_root_as_city_feature(&self.buffer.features_buf)?;
577        }
578        self.feat_no += 1;
579        self.cur_pos += feature_size as u64;
580
581        Ok(())
582    }
583
584    fn iter_get(&self) -> Option<&FcbBuffer> {
585        if self.state == State::Finished {
586            None
587        } else {
588            debug_assert!(self.state == State::Reading);
589            Some(&self.buffer)
590        }
591    }
592
593    fn iter_size_hint(&self) -> (usize, Option<usize>) {
594        if self.state == State::Finished {
595            (0, Some(0))
596        } else if let Some(count) = self.count {
597            let remaining = count - self.feat_no;
598            (remaining, Some(remaining))
599        } else {
600            (0, None)
601        }
602    }
603}
fcb_core/reader/mod.rs

fcb_core/reader/
mod.rs
