/// Copies a chunk in the Closed state into the ReadBuffer from
/// the mutable buffer and marks the chunk with `Moved` state
///
/// This code does not do any checking of the read buffer against
/// memory limits, etc
///
/// This (async) function returns when this process is complete,
/// but the process may take a long time
///
/// Returns a handle to the newly loaded chunk in the read buffer
Copies a chunk in the Closed state into the ReadBuffer from
the mutable buffer and marks the chunk with `Moved` state

This code does not do any checking of the read buffer against
memory limits, etc

This (async) function returns when this process is complete,
but the process may take a long time

Returns a handle to the newly loaded chunk in the read buffer
Copies a chunk in the Closed state into the ReadBuffer from the mutable buffer and marks the chunk with ` Moved ` state This code does not do any checking of the read buffer against memory limits etc This ( async ) function returns when this process is complete but the process may take a long time Returns a handle to the newly loaded chunk in the read buffer
pub async fn load_chunk_to_read_buffer(
        &self,
        partition_key: &str,
        table_name: &str,
        chunk_id: u32,
        tracker: &TaskRegistration,
    ) -> Result<Arc<DbChunk>> {
        let chunk = {
            let partition = self
                .preserved_catalog
                .state()
                .valid_partition(partition_key)
                .context(LoadingChunk {
                    partition_key,
                    table_name,
                    chunk_id,
                })?;
            let partition = partition.read();
            partition
                .chunk(table_name, chunk_id)
                .context(LoadingChunk {
                    partition_key,
                    table_name,
                    chunk_id,
                })?
        };
        // update the catalog to say we are processing this chunk and
        // then drop the lock while we do the work
        let (mb_chunk, table_summary) = {
            let mut chunk = chunk.write();
            let mb_chunk = chunk.set_moving(tracker).context(LoadingChunk {
                partition_key,
                table_name,
                chunk_id,
            })?;
            (mb_chunk, chunk.table_summary())
        };
        info!(%partition_key, %table_name, %chunk_id, "chunk marked MOVING, loading tables into read buffer");
        // create a new read buffer chunk with memory tracking
        let metrics = self
            .metrics_registry
            .register_domain_with_labels("read_buffer", self.metric_labels.clone());
        let mut rb_chunk = ReadBufferChunk::new(ReadBufferChunkMetrics::new(
            &metrics,
            self.preserved_catalog
                .state()
                .metrics()
                .memory()
                .read_buffer(),
        ));
        // load table into the new chunk one by one.
        debug!(%partition_key, %table_name, %chunk_id, table=%table_summary.name, "loading table to read buffer");
        let batch = mb_chunk
            .read_filter(Selection::All)
            // It is probably reasonable to recover from this error
            // (reset the chunk state to Open) but until that is
            // implemented (and tested) just panic
            .expect("Loading chunk to mutable buffer");
        let sorted = sort_record_batch(batch).expect("failed to sort");
        rb_chunk.upsert_table(&table_summary.name, sorted);
        // Relock the chunk again (nothing else should have been able
        // to modify the chunk state while we were moving it
        let mut chunk = chunk.write();
        // update the catalog to say we are done processing
        chunk.set_moved(Arc::new(rb_chunk)).context(LoadingChunk {
            partition_key,
            table_name,
            chunk_id,
        })?;
        debug!(%partition_key, %table_name, %chunk_id, "chunk marked MOVED. loading complete");
        Ok(DbChunk::snapshot(&chunk))
    }
pub async fn load_chunk_to_read_buffer ( & self , partition_key : & str , table_name : & str , chunk_id : u32 , tracker : & TaskRegistration , ) -> Result < Arc < DbChunk > > { let chunk = { let partition = self . preserved_catalog . state ( ) . valid_partition ( partition_key ) . context ( LoadingChunk { partition_key , table_name , chunk_id , } ) ? ; let partition = partition . read ( ) ; partition . chunk ( table_name , chunk_id ) . context ( LoadingChunk { partition_key , table_name , chunk_id , } ) ? } ; // update the catalog to say we are processing this chunk and // then drop the lock while we do the work let ( mb_chunk , table_summary ) = { let mut chunk = chunk . write ( ) ; let mb_chunk = chunk . set_moving ( tracker ) . context ( LoadingChunk { partition_key , table_name , chunk_id , } ) ? ; ( mb_chunk , chunk . table_summary ( ) ) } ; info ! ( partition_key table_name chunk_id " " ) ; // create a new read buffer chunk with memory tracking let metrics = self . metrics_registry . register_domain_with_labels ( " " , self . metric_labels . clone ( ) ) ; let mut rb_chunk = ReadBufferChunk :: new ( ReadBufferChunkMetrics :: new ( & metrics , self . preserved_catalog . state ( ) . metrics ( ) . memory ( ) . read_buffer ( ) , ) ) ; // load table into the new chunk one by one. debug ! ( partition_key table_name chunk_id table table_summary name " " ) ; let batch = mb_chunk . read_filter ( Selection :: All ) // It is probably reasonable to recover from this error // (reset the chunk state to Open) but until that is // implemented (and tested) just panic . expect ( " " ) ; let sorted = sort_record_batch ( batch ) . expect ( " " ) ; rb_chunk . upsert_table ( & table_summary . name , sorted ) ; // Relock the chunk again (nothing else should have been able // to modify the chunk state while we were moving it let mut chunk = chunk . write ( ) ; // update the catalog to say we are done processing chunk . set_moved ( Arc :: new ( rb_chunk ) ) . context ( LoadingChunk { partition_key , table_name , chunk_id , } ) ? ; debug ! ( partition_key table_name chunk_id " " ) ; Ok ( DbChunk :: snapshot ( & chunk ) ) }
Copies a chunk in the Closed state into the ReadBuffer from
the mutable buffer and marks the chunk with `Moved` state
Copies a chunk in the Closed state into the ReadBuffer from the mutable buffer and marks the chunk with ` Moved ` state
&self
partition_key
&str
table_name
&str
chunk_id
u32
tracker
&TaskRegistration