/// filter batches with maximum bytes to be send back consumer
filter batches with maximum bytes to be send back consumer
filter batches with maximum bytes to be send back consumer
pub fn filter(&self, slice: AsyncFileSlice, max_bytes: usize) -> Result<Batch, Error> {
        use std::os::unix::io::AsRawFd;
        let fd = slice.as_raw_fd();
        let mut batch_iterator =
            FileBatchIterator::new(fd, slice.position() as i64, slice.len() as i64);
        let mut filter_batch = Batch::<MemoryRecords>::default();
        filter_batch.base_offset = -1; // indicate this is unitialized
        filter_batch.set_offset_delta(-1); // make add_to_offset_delta correctly
        let mut total_bytes = 0;
        loop {
            if let Some(batch_result) = batch_iterator.next() {
                // we filter-map entire batches.  entire batches are process as group
                // if we can't fit current batch into max bytes then it is dicarded
                let file_batch = batch_result?;
                debug!(
                    current_batch_offset = file_batch.batch.base_offset,
                    current_batch_offset_delta = file_batch.offset_delta(),
                    filter_offset_delta = filter_batch.get_header().last_offset_delta,
                    filter_base_offset = filter_batch.base_offset,
                    filter_records = filter_batch.records().len(),
                    "starting filter processing"
                );
                let now = Instant::now();
                self.records_cb.clear();
                let array_ptr = self.instance.copy_memory_to(&file_batch.records)?;
                let filter_record_count = self
                    .filter_fn
                    .call((array_ptr as i32, file_batch.records.len() as i32))?;
                debug!(filter_record_count,filter_execution_time = %now.elapsed().as_millis());
                if filter_record_count == -1 {
                    return Err(anyhow!("filter failed"));
                }
                let bytes = self
                    .records_cb
                    .get()
                    .map(|m| m.copy_memory_from())
                    .unwrap_or_default();
                debug!(out_filter_bytes = bytes.len());
                // this is inefficient for now
                let mut records: MemoryRecords = vec![];
                records.decode(&mut Cursor::new(bytes), 0)?;
                // there are filtered records!!
                if records.is_empty() {
                    debug!("filters records empty");
                } else {
                    // set base offset if this is first time
                    if filter_batch.base_offset == -1 {
                        filter_batch.base_offset = file_batch.base_offset();
                    }
                    // difference between filter batch and and current batch
                    // since base are different we need update delta offset for each records
                    let relative_base_offset = filter_batch.base_offset - file_batch.base_offset();
                    for record in &mut records {
                        record.add_base_offset(relative_base_offset);
                    }
                    let record_bytes = records.write_size(0);
                    // if filter bytes exceed max bytes then we skip this batch
                    if total_bytes + record_bytes > max_bytes {
                        debug!(
                            total_bytes = total_bytes + record_bytes,
                            max_bytes, "total filter bytes reached"
                        );
                        return Ok(filter_batch);
                    }
                    total_bytes += record_bytes;
                    debug!(
                        filter_records = records.len(),
                        total_bytes, "finished filtering"
                    );
                    filter_batch.mut_records().append(&mut records);
                }
                // only increment filter offset delta if filter_batch has been initialized
                if filter_batch.base_offset != -1 {
                    debug!(
                        offset_delta = file_batch.offset_delta(),
                        "adding to offset delta"
                    );
                    filter_batch.add_to_offset_delta(file_batch.offset_delta() + 1);
                }
            } else {
                debug!(
                    total_records = filter_batch.records().len(),
                    "no more batches filter end"
                );
                return Ok(filter_batch);
            }
        }
    }
pub fn filter ( & self , slice : AsyncFileSlice , max_bytes : usize ) -> Result < Batch , Error > { use std :: os :: unix :: io :: AsRawFd ; let fd = slice . as_raw_fd ( ) ; let mut batch_iterator = FileBatchIterator :: new ( fd , slice . position ( ) as i64 , slice . len ( ) as i64 ) ; let mut filter_batch = Batch :: < MemoryRecords > :: default ( ) ; filter_batch . base_offset = - 1 ; // indicate this is unitialized filter_batch . set_offset_delta ( - 1 ) ; // make add_to_offset_delta correctly let mut total_bytes = 0 ; loop { if let Some ( batch_result ) = batch_iterator . next ( ) { // we filter-map entire batches.  entire batches are process as group // if we can't fit current batch into max bytes then it is dicarded let file_batch = batch_result ? ; debug ! ( current_batch_offset file_batch batch base_offset current_batch_offset_delta file_batch offset_delta ( ) filter_offset_delta filter_batch get_header ( ) last_offset_delta filter_base_offset filter_batch base_offset filter_records filter_batch records ( ) len ( ) " " ) ; let now = Instant :: now ( ) ; self . records_cb . clear ( ) ; let array_ptr = self . instance . copy_memory_to ( & file_batch . records ) ? ; let filter_record_count = self . filter_fn . call ( ( array_ptr as i32 , file_batch . records . len ( ) as i32 ) ) ? ; debug ! ( filter_record_count filter_execution_time now elapsed ( ) as_millis ( ) ) ; if filter_record_count == - 1 { return Err ( anyhow ! ( " " ) ) ; } let bytes = self . records_cb . get ( ) . map ( | m | m . copy_memory_from ( ) ) . unwrap_or_default ( ) ; debug ! ( out_filter_bytes bytes len ( ) ) ; // this is inefficient for now let mut records : MemoryRecords = vec ! [ ] ; records . decode ( & mut Cursor :: new ( bytes ) , 0 ) ? ; // there are filtered records!! if records . is_empty ( ) { debug ! ( " " ) ; } else { // set base offset if this is first time if filter_batch . base_offset == - 1 { filter_batch . base_offset = file_batch . base_offset ( ) ; } // difference between filter batch and and current batch // since base are different we need update delta offset for each records let relative_base_offset = filter_batch . base_offset - file_batch . base_offset ( ) ; for record in & mut records { record . add_base_offset ( relative_base_offset ) ; } let record_bytes = records . write_size ( 0 ) ; // if filter bytes exceed max bytes then we skip this batch if total_bytes + record_bytes > max_bytes { debug ! ( total_bytes total_bytes record_bytes max_bytes " " ) ; return Ok ( filter_batch ) ; } total_bytes += record_bytes ; debug ! ( filter_records records len ( ) total_bytes " " ) ; filter_batch . mut_records ( ) . append ( & mut records ) ; } // only increment filter offset delta if filter_batch has been initialized if filter_batch . base_offset != - 1 { debug ! ( offset_delta file_batch offset_delta ( ) " " ) ; filter_batch . add_to_offset_delta ( file_batch . offset_delta ( ) + 1 ) ; } } else { debug ! ( total_records filter_batch records ( ) len ( ) " " ) ; return Ok ( filter_batch ) ; } } }
filter batches with maximum bytes to be send back consumer
filter batches with maximum bytes to be send back consumer
&self
slice
AsyncFileSlice
max_bytes
usize