fn collect_finished_data(&mut self) -> Result<Vec<Row>, String> {
        // Check if there exist any errors and, if so, return whatever one we
        // find first.
        let (mut cursor, storage) = self.trace_bundle.errs_mut().cursor();
        while cursor.key_valid(&storage) {
            let mut copies = 0;
            cursor.map_times(&storage, |time, diff| {
                if time.less_equal(&self.timestamp) {
                    copies += diff;
                }
            });
            if copies < 0 {
                return Err(format!(
                    "Invalid data in source errors, saw retractions ({}) for row that does not exist: {}",
                    copies * -1,
                    cursor.key(&storage),
                ));
            }
            if copies > 0 {
                return Err(cursor.key(&storage).to_string());
            }
            cursor.step_key(&storage);
        }

        // Cursor and bound lifetime for `Row` data in the backing trace.
        let (mut cursor, storage) = self.trace_bundle.oks_mut().cursor();
        // Accumulated `Vec<Datum>` results that we are likely to return.
        let mut results = Vec::new();

        // When set, a bound on the number of records we need to return.
        // The requirements on the records are driven by the finishing's
        // `order_by` field. Further limiting will happen when the results
        // are collected, so we don't need to have exactly this many results,
        // just at least those results that would have been returned.
        let max_results = self.finishing.limit.map(|l| l + self.finishing.offset);

        if let Some(literal) = &self.key {
            cursor.seek_key(&storage, literal);
        }

        let mut row_builder = Row::default();
        let mut datum_vec = DatumVec::new();
        let mut l_datum_vec = DatumVec::new();
        let mut r_datum_vec = DatumVec::new();

        while cursor.key_valid(&storage) {
            while cursor.val_valid(&storage) {
                // TODO: This arena could be maintained and reuse for longer
                // but it wasn't clear at what granularity we should flush
                // it to ensure we don't accidentally spike our memory use.
                // This choice is conservative, and not the end of the world
                // from a performance perspective.
                let arena = RowArena::new();
                let key = cursor.key(&storage);
                let row = cursor.val(&storage);
                // TODO: We could unpack into a re-used allocation, except
                // for the arena above (the allocation would not be allowed
                // to outlive the arena above, from which it might borrow).
                let mut borrow = datum_vec.borrow_with_many(&[key, row]);
                if let Some(result) = self
                    .map_filter_project
                    .evaluate_into(&mut borrow, &arena, &mut row_builder)
                    .map_err_to_string()?
                {
                    let mut copies = 0;
                    cursor.map_times(&storage, |time, diff| {
                        if time.less_equal(&self.timestamp) {
                            copies += diff;
                        }
                    });
                    if copies < 0 {
                        return Err(format!(
                            "Invalid data in source, saw retractions ({}) for row that does not exist: {:?}",
                            copies * -1,
                            &*borrow,
                        ));
                    }

                    // When we have a LIMIT we can restrict the number of copies we make.
                    // This protects us when we have many copies of the same records, as
                    // the DD representation uses a binary count and may not exhaust our
                    // memory in situtations where this copying might.
                    if let Some(limit) = max_results {
                        let limit = std::convert::TryInto::<isize>::try_into(limit);
                        if let Ok(limit) = limit {
                            copies = std::cmp::min(copies, limit);
                        }
                    }
                    for _ in 0..copies {
                        results.push(result.clone());
                    }

                    // If we hold many more than `max_results` records, we can thin down
                    // `results` using `self.finishing.ordering`.
                    if let Some(max_results) = max_results {
                        // We use a threshold twice what we intend, to amortize the work
                        // across all of the insertions. We could tighten this, but it
                        // works for the moment.
                        if results.len() >= 2 * max_results {
                            if self.finishing.order_by.is_empty() {
                                results.truncate(max_results);
                                return Ok(results);
                            } else {
                                // We can sort `results` and then truncate to `max_results`.
                                // This has an effect similar to a priority queue, without
                                // its interactive dequeueing properties.
                                // TODO: Had we left these as `Vec<Datum>` we would avoid
                                // the unpacking; we should consider doing that, although
                                // it will require a re-pivot of the code to branch on this
                                // inner test (as we prefer not to maintain `Vec<Datum>`
                                // in the other case).
                                results.sort_by(|left, right| {
                                    let left_datums = l_datum_vec.borrow_with(left);
                                    let right_datums = r_datum_vec.borrow_with(right);
                                    expr::compare_columns(
                                        &self.finishing.order_by,
                                        &left_datums,
                                        &right_datums,
                                        || left.cmp(&right),
                                    )
                                });
                                results.truncate(max_results);
                            }
                        }
                    }
                }
                cursor.step_val(&storage);
            }
            // If we had a key, we are now done and can return.
            if self.key.is_some() {
                return Ok(results);
            } else {
                cursor.step_key(&storage);
            }
        }

        Ok(results)
    }
/// Collects data for a known-complete peek.
Collects data for a known-complete peek.
Collects data for a known - complete peek .
fn collect_finished_data(&mut self) -> Result<Vec<Row>, String> {
        // Check if there exist any errors and, if so, return whatever one we
        // find first.
        let (mut cursor, storage) = self.trace_bundle.errs_mut().cursor();
        while cursor.key_valid(&storage) {
            let mut copies = 0;
            cursor.map_times(&storage, |time, diff| {
                if time.less_equal(&self.timestamp) {
                    copies += diff;
                }
            });
            if copies < 0 {
                return Err(format!(
                    "Invalid data in source errors, saw retractions ({}) for row that does not exist: {}",
                    copies * -1,
                    cursor.key(&storage),
                ));
            }
            if copies > 0 {
                return Err(cursor.key(&storage).to_string());
            }
            cursor.step_key(&storage);
        }
        // Cursor and bound lifetime for `Row` data in the backing trace.
        let (mut cursor, storage) = self.trace_bundle.oks_mut().cursor();
        // Accumulated `Vec<Datum>` results that we are likely to return.
        let mut results = Vec::new();
        // When set, a bound on the number of records we need to return.
        // The requirements on the records are driven by the finishing's
        // `order_by` field. Further limiting will happen when the results
        // are collected, so we don't need to have exactly this many results,
        // just at least those results that would have been returned.
        let max_results = self.finishing.limit.map(|l| l + self.finishing.offset);
        if let Some(literal) = &self.key {
            cursor.seek_key(&storage, literal);
        }
        let mut row_builder = Row::default();
        let mut datum_vec = DatumVec::new();
        let mut l_datum_vec = DatumVec::new();
        let mut r_datum_vec = DatumVec::new();
        while cursor.key_valid(&storage) {
            while cursor.val_valid(&storage) {
                // TODO: This arena could be maintained and reuse for longer
                // but it wasn't clear at what granularity we should flush
                // it to ensure we don't accidentally spike our memory use.
                // This choice is conservative, and not the end of the world
                // from a performance perspective.
                let arena = RowArena::new();
                let key = cursor.key(&storage);
                let row = cursor.val(&storage);
                // TODO: We could unpack into a re-used allocation, except
                // for the arena above (the allocation would not be allowed
                // to outlive the arena above, from which it might borrow).
                let mut borrow = datum_vec.borrow_with_many(&[key, row]);
                if let Some(result) = self
                    .map_filter_project
                    .evaluate_into(&mut borrow, &arena, &mut row_builder)
                    .map_err_to_string()?
                {
                    let mut copies = 0;
                    cursor.map_times(&storage, |time, diff| {
                        if time.less_equal(&self.timestamp) {
                            copies += diff;
                        }
                    });
                    if copies < 0 {
                        return Err(format!(
                            "Invalid data in source, saw retractions ({}) for row that does not exist: {:?}",
                            copies * -1,
                            &*borrow,
                        ));
                    }
                    // When we have a LIMIT we can restrict the number of copies we make.
                    // This protects us when we have many copies of the same records, as
                    // the DD representation uses a binary count and may not exhaust our
                    // memory in situtations where this copying might.
                    if let Some(limit) = max_results {
                        let limit = std::convert::TryInto::<isize>::try_into(limit);
                        if let Ok(limit) = limit {
                            copies = std::cmp::min(copies, limit);
                        }
                    }
                    for _ in 0..copies {
                        results.push(result.clone());
                    }
                    // If we hold many more than `max_results` records, we can thin down
                    // `results` using `self.finishing.ordering`.
                    if let Some(max_results) = max_results {
                        // We use a threshold twice what we intend, to amortize the work
                        // across all of the insertions. We could tighten this, but it
                        // works for the moment.
                        if results.len() >= 2 * max_results {
                            if self.finishing.order_by.is_empty() {
                                results.truncate(max_results);
                                return Ok(results);
                            } else {
                                // We can sort `results` and then truncate to `max_results`.
                                // This has an effect similar to a priority queue, without
                                // its interactive dequeueing properties.
                                // TODO: Had we left these as `Vec<Datum>` we would avoid
                                // the unpacking; we should consider doing that, although
                                // it will require a re-pivot of the code to branch on this
                                // inner test (as we prefer not to maintain `Vec<Datum>`
                                // in the other case).
                                results.sort_by(|left, right| {
                                    let left_datums = l_datum_vec.borrow_with(left);
                                    let right_datums = r_datum_vec.borrow_with(right);
                                    expr::compare_columns(
                                        &self.finishing.order_by,
                                        &left_datums,
                                        &right_datums,
                                        || left.cmp(&right),
                                    )
                                });
                                results.truncate(max_results);
                            }
                        }
                    }
                }
                cursor.step_val(&storage);
            }
            // If we had a key, we are now done and can return.
            if self.key.is_some() {
                return Ok(results);
            } else {
                cursor.step_key(&storage);
            }
        }
        Ok(results)
    }
fn collect_finished_data ( & mut self ) -> Result < Vec < Row > , String > { // Check if there exist any errors and, if so, return whatever one we // find first. let ( mut cursor , storage ) = self . trace_bundle . errs_mut ( ) . cursor ( ) ; while cursor . key_valid ( & storage ) { let mut copies = 0 ; cursor . map_times ( & storage , | time , diff | { if time . less_equal ( & self . timestamp ) { copies += diff ; } } ) ; if copies < 0 { return Err ( format ! ( " " copies 1 cursor key ( storage ) ) ) ; } if copies > 0 { return Err ( cursor . key ( & storage ) . to_string ( ) ) ; } cursor . step_key ( & storage ) ; } // Cursor and bound lifetime for `Row` data in the backing trace. let ( mut cursor , storage ) = self . trace_bundle . oks_mut ( ) . cursor ( ) ; // Accumulated `Vec<Datum>` results that we are likely to return. let mut results = Vec :: new ( ) ; // When set, a bound on the number of records we need to return. // The requirements on the records are driven by the finishing's // `order_by` field. Further limiting will happen when the results // are collected, so we don't need to have exactly this many results, // just at least those results that would have been returned. let max_results = self . finishing . limit . map ( | l | l + self . finishing . offset ) ; if let Some ( literal ) = & self . key { cursor . seek_key ( & storage , literal ) ; } let mut row_builder = Row :: default ( ) ; let mut datum_vec = DatumVec :: new ( ) ; let mut l_datum_vec = DatumVec :: new ( ) ; let mut r_datum_vec = DatumVec :: new ( ) ; while cursor . key_valid ( & storage ) { while cursor . val_valid ( & storage ) { // TODO: This arena could be maintained and reuse for longer // but it wasn't clear at what granularity we should flush // it to ensure we don't accidentally spike our memory use. // This choice is conservative, and not the end of the world // from a performance perspective. let arena = RowArena :: new ( ) ; let key = cursor . key ( & storage ) ; let row = cursor . val ( & storage ) ; // TODO: We could unpack into a re-used allocation, except // for the arena above (the allocation would not be allowed // to outlive the arena above, from which it might borrow). let mut borrow = datum_vec . borrow_with_many ( & [ key , row ] ) ; if let Some ( result ) = self . map_filter_project . evaluate_into ( & mut borrow , & arena , & mut row_builder ) . map_err_to_string ( ) ? { let mut copies = 0 ; cursor . map_times ( & storage , | time , diff | { if time . less_equal ( & self . timestamp ) { copies += diff ; } } ) ; if copies < 0 { return Err ( format ! ( " " copies 1 borrow ) ) ; } // When we have a LIMIT we can restrict the number of copies we make. // This protects us when we have many copies of the same records, as // the DD representation uses a binary count and may not exhaust our // memory in situtations where this copying might. if let Some ( limit ) = max_results { let limit = std :: convert :: TryInto :: < isize > :: try_into ( limit ) ; if let Ok ( limit ) = limit { copies = std :: cmp :: min ( copies , limit ) ; } } for _ in 0 .. copies { results . push ( result . clone ( ) ) ; } // If we hold many more than `max_results` records, we can thin down // `results` using `self.finishing.ordering`. if let Some ( max_results ) = max_results { // We use a threshold twice what we intend, to amortize the work // across all of the insertions. We could tighten this, but it // works for the moment. if results . len ( ) >= 2 * max_results { if self . finishing . order_by . is_empty ( ) { results . truncate ( max_results ) ; return Ok ( results ) ; } else { // We can sort `results` and then truncate to `max_results`. // This has an effect similar to a priority queue, without // its interactive dequeueing properties. // TODO: Had we left these as `Vec<Datum>` we would avoid // the unpacking; we should consider doing that, although // it will require a re-pivot of the code to branch on this // inner test (as we prefer not to maintain `Vec<Datum>` // in the other case). results . sort_by ( | left , right | { let left_datums = l_datum_vec . borrow_with ( left ) ; let right_datums = r_datum_vec . borrow_with ( right ) ; expr :: compare_columns ( & self . finishing . order_by , & left_datums , & right_datums , | | left . cmp ( & right ) , ) } ) ; results . truncate ( max_results ) ; } } } } cursor . step_val ( & storage ) ; } // If we had a key, we are now done and can return. if self . key . is_some ( ) { return Ok ( results ) ; } else { cursor . step_key ( & storage ) ; } } Ok ( results ) }
Collects data for a known-complete peek.
Collects data for a known - complete peek .
&mut self
&mut self