Method normalize

src/persist-client/src/fetch.rs:1406–1486

Returns the updates with all truncation / timestamp rewriting applied.

(&self, metrics: &ColumnarMetrics)

    /// Returns the updates with all truncation / timestamp rewriting applied.
    pub(crate) fn normalize(&self, metrics: &ColumnarMetrics) -> BlobTraceUpdates {
        let updates = self.part.updates.clone();
        if !self.needs_truncation && self.ts_rewrite.is_none() {
            return updates;
        }

        let mut codec = updates
            .records()
            .map(|r| (r.keys().clone(), r.vals().clone()));
        let mut structured = updates.structured().cloned();
        let mut timestamps = updates.timestamps().clone();
        let mut diffs = updates.diffs().clone();

        if let Some(rewrite) = self.ts_rewrite.as_ref() {
            timestamps = arrow::compute::unary(&timestamps, |i: i64| {
                let mut t = T::decode(i.to_le_bytes());
                t.advance_by(rewrite.borrow());
                i64::from_le_bytes(T::encode(&t))
            });
        }

        let reallocated = if self.needs_truncation {
            let filter = BooleanArray::from_unary(&timestamps, |i| {
                let t = T::decode(i.to_le_bytes());
                let truncate_t = {
                    !self.registered_desc.lower().less_equal(&t)
                        || self.registered_desc.upper().less_equal(&t)
                };
                !truncate_t
            });
            if filter.false_count() == 0 {
                // If we're not filtering anything in practice, skip filtering and reallocating.
                false
            } else {
                let filter = FilterBuilder::new(&filter).optimize().build();
                let do_filter = |array: &dyn Array| filter.filter(array).expect("valid filter len");
                if let Some((keys, vals)) = codec {
                    codec = Some((
                        realloc_array(do_filter(&keys).as_binary(), metrics),
                        realloc_array(do_filter(&vals).as_binary(), metrics),
                    ));
                }
                if let Some(ext) = structured {
                    structured = Some(ColumnarRecordsStructuredExt {
                        key: realloc_any(do_filter(&*ext.key), metrics),
                        val: realloc_any(do_filter(&*ext.val), metrics),
                    });
                }
                timestamps = realloc_array(do_filter(&timestamps).as_primitive(), metrics);
                diffs = realloc_array(do_filter(&diffs).as_primitive(), metrics);
                true
            }
        } else {
            false
        };

        if self.ts_rewrite.is_some() && !reallocated {
            timestamps = realloc_array(&timestamps, metrics);