/// Add a record batch of data to to a `Table` in the chunk.
///
/// The data is converted to a `RowGroup` outside of any locking on the
/// `Chunk` so the caller does not need to be concerned about the size of
/// the update. If the `Table` already exists then a new `RowGroup` will be
/// added to the `Table`. Otherwise a new `Table` with a single `RowGroup`
/// will be created.
Add a record batch of data to to a `Table` in the chunk.
The data is converted to a `RowGroup` outside of any locking on the
`Chunk` so the caller does not need to be concerned about the size of
the update.
Add a record batch of data to to a ` Table ` in the chunk . The data is converted to a ` RowGroup ` outside of any locking on the ` Chunk ` so the caller does not need to be concerned about the size of the update .
pub fn upsert_table(&mut self, table_name: impl Into<String>, table_data: RecordBatch) {
        // Approximate heap size of record batch.
        let mub_rb_size = table_data
            .columns()
            .iter()
            .map(|c| c.get_buffer_memory_size())
            .sum::<usize>();
        let columns = table_data.num_columns();
        // This call is expensive. Complete it before locking.
        let now = std::time::Instant::now();
        let row_group = RowGroup::from(table_data);
        let compressing_took = now.elapsed();
        let table_name = table_name.into();
        let rows = row_group.rows();
        let rg_size = row_group.size();
        let mub_rb_comp = format!(
            "{:.2}%",
            (1.0 - (rg_size as f64 / mub_rb_size as f64)) * 100.0
        );
        let raw_size_null = row_group.size_raw(true);
        let raw_size_no_null = row_group.size_raw(false);
        let raw_rb_comp = format!(
            "{:.2}%",
            (1.0 - (rg_size as f64 / raw_size_null as f64)) * 100.0
        );
        info!(%rows, %columns, rg_size, mub_rb_size, %mub_rb_comp, raw_size_null, raw_size_no_null, %raw_rb_comp, ?table_name, ?compressing_took, "row group added");
        let mut chunk_data = self.chunk_data.write();
        // update the meta-data for this chunk with contents of row group.
        chunk_data.rows += row_group.rows() as u64;
        chunk_data.row_groups += 1;
        // track new row group statistics to update column-based metrics.
        let storage_statistics = row_group.column_storage_statistics();
        // create a new table if one doesn't exist, or add the table data to
        // the existing table.
        match chunk_data.data.entry(table_name.clone()) {
            Entry::Occupied(mut table_entry) => {
                let table = table_entry.get_mut();
                table.add_row_group(row_group);
            }
            Entry::Vacant(table_entry) => {
                // add a new table to this chunk.
                table_entry.insert(Table::new(table_name, row_group));
            }
        };
        // Get and set new size of chunk on memory tracker
        let size = Self::base_size() + chunk_data.size();
        self.metrics.memory_bytes.set(size);
        // update column metrics associated with column storage
        std::mem::drop(chunk_data); // drop write lock
        self.metrics
            .update_column_storage_statistics(&storage_statistics);
    }
pub fn upsert_table ( & mut self , table_name : impl Into < String > , table_data : RecordBatch ) { // Approximate heap size of record batch. let mub_rb_size = table_data . columns ( ) . iter ( ) . map ( | c | c . get_buffer_memory_size ( ) ) . sum :: < usize > ( ) ; let columns = table_data . num_columns ( ) ; // This call is expensive. Complete it before locking. let now = std :: time :: Instant :: now ( ) ; let row_group = RowGroup :: from ( table_data ) ; let compressing_took = now . elapsed ( ) ; let table_name = table_name . into ( ) ; let rows = row_group . rows ( ) ; let rg_size = row_group . size ( ) ; let mub_rb_comp = format ! ( " " ( 1.0 ( rg_size as f64 mub_rb_size as f64 ) ) 100.0 ) ; let raw_size_null = row_group . size_raw ( true ) ; let raw_size_no_null = row_group . size_raw ( false ) ; let raw_rb_comp = format ! ( " " ( 1.0 ( rg_size as f64 raw_size_null as f64 ) ) 100.0 ) ; info ! ( rows columns rg_size mub_rb_size mub_rb_comp raw_size_null raw_size_no_null raw_rb_comp table_name compressing_took " " ) ; let mut chunk_data = self . chunk_data . write ( ) ; // update the meta-data for this chunk with contents of row group. chunk_data . rows += row_group . rows ( ) as u64 ; chunk_data . row_groups += 1 ; // track new row group statistics to update column-based metrics. let storage_statistics = row_group . column_storage_statistics ( ) ; // create a new table if one doesn't exist, or add the table data to // the existing table. match chunk_data . data . entry ( table_name . clone ( ) ) { Entry :: Occupied ( mut table_entry ) => { let table = table_entry . get_mut ( ) ; table . add_row_group ( row_group ) ; } Entry :: Vacant ( table_entry ) => { // add a new table to this chunk. table_entry . insert ( Table :: new ( table_name , row_group ) ) ; } } ; // Get and set new size of chunk on memory tracker let size = Self :: base_size ( ) + chunk_data . size ( ) ; self . metrics . memory_bytes . set ( size ) ; // update column metrics associated with column storage std :: mem :: drop ( chunk_data ) ; // drop write lock self . metrics . update_column_storage_statistics ( & storage_statistics ) ; }
Add a record batch of data to to a `Table` in the chunk.
Add a record batch of data to to a ` Table ` in the chunk .
&mut self
table_name
impl Into<String>
table_data
RecordBatch