pub(crate) fn compact_chunks(
    partition: LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>,
    chunks: Vec<LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>>,
) -> Result<(
    TaskTracker<Job>,
    TrackedFuture<impl Future<Output = Result<Arc<DbChunk>>> + Send>,
)> {
    let now = std::time::Instant::now(); // time compaction duration.
    let db = Arc::clone(&partition.data().db);
    let table_name = partition.table_name().to_string();
    let chunk_ids: Vec<_> = chunks.iter().map(|x| x.id()).collect();

    let (tracker, registration) = db.jobs.register(Job::CompactChunks {
        partition: partition.addr().clone(),
        chunks: chunk_ids.clone(),
    });

    // Mark and snapshot chunks, then drop locks
    let mut input_rows = 0;
    let mut time_of_first_write: Option<DateTime<Utc>> = None;
    let mut time_of_last_write: Option<DateTime<Utc>> = None;
    let query_chunks = chunks
        .into_iter()
        .map(|mut chunk| {
            // Sanity-check
            assert!(Arc::ptr_eq(&db, &chunk.data().db));
            assert_eq!(chunk.table_name().as_ref(), table_name.as_str());

            input_rows += chunk.table_summary().total_count();

            let candidate_first = chunk.time_of_first_write();
            time_of_first_write = time_of_first_write
                .map(|prev_first| prev_first.min(candidate_first))
                .or(Some(candidate_first));

            let candidate_last = chunk.time_of_last_write();
            time_of_last_write = time_of_last_write
                .map(|prev_last| prev_last.max(candidate_last))
                .or(Some(candidate_last));

            chunk.set_compacting(&registration)?;
            Ok(DbChunk::snapshot(&*chunk))
        })
        .collect::<Result<Vec<_>>>()?;

    // drop partition lock
    let partition = partition.into_data().partition;

    let time_of_first_write = time_of_first_write.expect("Should have had a first write somewhere");
    let time_of_last_write = time_of_last_write.expect("Should have had a last write somewhere");

    let ctx = db.exec.new_context(ExecutorType::Reorg);

    let fut = async move {
        let fut_now = std::time::Instant::now();
        let key = compute_sort_key(query_chunks.iter().map(|x| x.summary()));
        let key_str = format!("\"{}\"", key); // for logging

        // build schema
        //
        // Note: we only use the merged schema from the to-be-compacted
        // chunks - not the table-wide schema, since we don't need to
        // bother with other columns (e.g. ones that only exist in other
        // partitions).
        let schema = merge_schemas(&query_chunks);

        // Cannot move query_chunks as the sort key borrows the column names
        let (schema, plan) =
            ReorgPlanner::new().compact_plan(schema, query_chunks.iter().map(Arc::clone), key)?;

        let physical_plan = ctx.prepare_plan(&plan)?;
        let stream = ctx.execute(physical_plan).await?;
        let rb_chunk = collect_rub(stream, &db, &table_name)
            .await?
            .expect("chunk has zero rows");
        let rb_row_groups = rb_chunk.row_groups();

        let new_chunk = {
            let mut partition = partition.write();
            for id in chunk_ids {
                partition.force_drop_chunk(id)
            }
            partition.create_rub_chunk(rb_chunk, time_of_first_write, time_of_last_write, schema)
        };

        let guard = new_chunk.read();
        let elapsed = now.elapsed();

        assert!(
            guard.table_summary().total_count() > 0,
            "chunk has zero rows"
        );
        // input rows per second
        let throughput = (input_rows as u128 * 1_000_000_000) / elapsed.as_nanos();

        info!(input_chunks=query_chunks.len(), rub_row_groups=rb_row_groups,
                input_rows=input_rows, output_rows=guard.table_summary().total_count(),
                sort_key=%key_str, compaction_took = ?elapsed, fut_execution_duration= ?fut_now.elapsed(),
                rows_per_sec=?throughput,  "chunk(s) compacted");

        Ok(DbChunk::snapshot(&guard))
    };

    Ok((tracker, fut.track(registration)))
}
/// Compact the provided chunks into a single chunk,
/// returning the newly created chunk
///
/// TODO: Replace low-level locks with transaction object
Compact the provided chunks into a single chunk,
returning the newly created chunk

Replace low-level locks with transaction object
Compact the provided chunks into a single chunk returning the newly created chunk Replace low - level locks with transaction object
pub(crate) fn compact_chunks(
    partition: LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>,
    chunks: Vec<LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>>,
) -> Result<(
    TaskTracker<Job>,
    TrackedFuture<impl Future<Output = Result<Arc<DbChunk>>> + Send>,
)> {
    let now = std::time::Instant::now(); // time compaction duration.
    let db = Arc::clone(&partition.data().db);
    let table_name = partition.table_name().to_string();
    let chunk_ids: Vec<_> = chunks.iter().map(|x| x.id()).collect();
    let (tracker, registration) = db.jobs.register(Job::CompactChunks {
        partition: partition.addr().clone(),
        chunks: chunk_ids.clone(),
    });
    // Mark and snapshot chunks, then drop locks
    let mut input_rows = 0;
    let mut time_of_first_write: Option<DateTime<Utc>> = None;
    let mut time_of_last_write: Option<DateTime<Utc>> = None;
    let query_chunks = chunks
        .into_iter()
        .map(|mut chunk| {
            // Sanity-check
            assert!(Arc::ptr_eq(&db, &chunk.data().db));
            assert_eq!(chunk.table_name().as_ref(), table_name.as_str());
            input_rows += chunk.table_summary().total_count();
            let candidate_first = chunk.time_of_first_write();
            time_of_first_write = time_of_first_write
                .map(|prev_first| prev_first.min(candidate_first))
                .or(Some(candidate_first));
            let candidate_last = chunk.time_of_last_write();
            time_of_last_write = time_of_last_write
                .map(|prev_last| prev_last.max(candidate_last))
                .or(Some(candidate_last));
            chunk.set_compacting(&registration)?;
            Ok(DbChunk::snapshot(&*chunk))
        })
        .collect::<Result<Vec<_>>>()?;
    // drop partition lock
    let partition = partition.into_data().partition;
    let time_of_first_write = time_of_first_write.expect("Should have had a first write somewhere");
    let time_of_last_write = time_of_last_write.expect("Should have had a last write somewhere");
    let ctx = db.exec.new_context(ExecutorType::Reorg);
    let fut = async move {
        let fut_now = std::time::Instant::now();
        let key = compute_sort_key(query_chunks.iter().map(|x| x.summary()));
        let key_str = format!("\"{}\"", key); // for logging
        // build schema
        //
        // Note: we only use the merged schema from the to-be-compacted
        // chunks - not the table-wide schema, since we don't need to
        // bother with other columns (e.g. ones that only exist in other
        // partitions).
        let schema = merge_schemas(&query_chunks);
        // Cannot move query_chunks as the sort key borrows the column names
        let (schema, plan) =
            ReorgPlanner::new().compact_plan(schema, query_chunks.iter().map(Arc::clone), key)?;
        let physical_plan = ctx.prepare_plan(&plan)?;
        let stream = ctx.execute(physical_plan).await?;
        let rb_chunk = collect_rub(stream, &db, &table_name)
            .await?
            .expect("chunk has zero rows");
        let rb_row_groups = rb_chunk.row_groups();
        let new_chunk = {
            let mut partition = partition.write();
            for id in chunk_ids {
                partition.force_drop_chunk(id)
            }
            partition.create_rub_chunk(rb_chunk, time_of_first_write, time_of_last_write, schema)
        };
        let guard = new_chunk.read();
        let elapsed = now.elapsed();
        assert!(
            guard.table_summary().total_count() > 0,
            "chunk has zero rows"
        );
        // input rows per second
        let throughput = (input_rows as u128 * 1_000_000_000) / elapsed.as_nanos();
        info!(input_chunks=query_chunks.len(), rub_row_groups=rb_row_groups,
                input_rows=input_rows, output_rows=guard.table_summary().total_count(),
                sort_key=%key_str, compaction_took = ?elapsed, fut_execution_duration= ?fut_now.elapsed(),
                rows_per_sec=?throughput,  "chunk(s) compacted");
        Ok(DbChunk::snapshot(&guard))
    };
    Ok((tracker, fut.track(registration)))
}
pub ( crate ) fn compact_chunks ( partition : LifecycleWriteGuard < ' _ , Partition , LockableCatalogPartition > , chunks : Vec < LifecycleWriteGuard < ' _ , CatalogChunk , LockableCatalogChunk > > , ) -> Result < ( TaskTracker < Job > , TrackedFuture < impl Future < Output = Result < Arc < DbChunk > > > + Send > , ) > { let now = std :: time :: Instant :: now ( ) ; // time compaction duration. let db = Arc :: clone ( & partition . data ( ) . db ) ; let table_name = partition . table_name ( ) . to_string ( ) ; let chunk_ids : Vec < _ > = chunks . iter ( ) . map ( | x | x . id ( ) ) . collect ( ) ; let ( tracker , registration ) = db . jobs . register ( Job :: CompactChunks { partition : partition . addr ( ) . clone ( ) , chunks : chunk_ids . clone ( ) , } ) ; // Mark and snapshot chunks, then drop locks let mut input_rows = 0 ; let mut time_of_first_write : Option < DateTime < Utc > > = None ; let mut time_of_last_write : Option < DateTime < Utc > > = None ; let query_chunks = chunks . into_iter ( ) . map ( | mut chunk | { // Sanity-check assert ! ( Arc ptr_eq ( db chunk data ( ) db ) ) ; assert_eq ! ( chunk table_name ( ) as_ref ( ) table_name as_str ( ) ) ; input_rows += chunk . table_summary ( ) . total_count ( ) ; let candidate_first = chunk . time_of_first_write ( ) ; time_of_first_write = time_of_first_write . map ( | prev_first | prev_first . min ( candidate_first ) ) . or ( Some ( candidate_first ) ) ; let candidate_last = chunk . time_of_last_write ( ) ; time_of_last_write = time_of_last_write . map ( | prev_last | prev_last . max ( candidate_last ) ) . or ( Some ( candidate_last ) ) ; chunk . set_compacting ( & registration ) ? ; Ok ( DbChunk :: snapshot ( & * chunk ) ) } ) . collect :: < Result < Vec < _ > > > ( ) ? ; // drop partition lock let partition = partition . into_data ( ) . partition ; let time_of_first_write = time_of_first_write . expect ( " " ) ; let time_of_last_write = time_of_last_write . expect ( " " ) ; let ctx = db . exec . new_context ( ExecutorType :: Reorg ) ; let fut = async move { let fut_now = std :: time :: Instant :: now ( ) ; let key = compute_sort_key ( query_chunks . iter ( ) . map ( | x | x . summary ( ) ) ) ; let key_str = format ! ( " \" \" " key ) ; // for logging // build schema // // Note: we only use the merged schema from the to-be-compacted // chunks - not the table-wide schema, since we don't need to // bother with other columns (e.g. ones that only exist in other // partitions). let schema = merge_schemas ( & query_chunks ) ; // Cannot move query_chunks as the sort key borrows the column names let ( schema , plan ) = ReorgPlanner :: new ( ) . compact_plan ( schema , query_chunks . iter ( ) . map ( Arc :: clone ) , key ) ? ; let physical_plan = ctx . prepare_plan ( & plan ) ? ; let stream = ctx . execute ( physical_plan ) . await ? ; let rb_chunk = collect_rub ( stream , & db , & table_name ) . await ? . expect ( " " ) ; let rb_row_groups = rb_chunk . row_groups ( ) ; let new_chunk = { let mut partition = partition . write ( ) ; for id in chunk_ids { partition . force_drop_chunk ( id ) } partition . create_rub_chunk ( rb_chunk , time_of_first_write , time_of_last_write , schema ) } ; let guard = new_chunk . read ( ) ; let elapsed = now . elapsed ( ) ; assert ! ( guard table_summary ( ) total_count ( ) 0 " " ) ; // input rows per second let throughput = ( input_rows as u128 * 1_000_000_000 ) / elapsed . as_nanos ( ) ; info ! ( input_chunks query_chunks len ( ) rub_row_groups rb_row_groups input_rows input_rows output_rows guard table_summary ( ) total_count ( ) sort_key key_str compaction_took elapsed fut_execution_duration fut_now elapsed ( ) rows_per_sec throughput " " ) ; Ok ( DbChunk :: snapshot ( & guard ) ) } ; Ok ( ( tracker , fut . track ( registration ) ) ) }
Compact the provided chunks into a single chunk,
returning the newly created chunk
Compact the provided chunks into a single chunk returning the newly created chunk
partition
LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>
chunks
Vec<LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>>
partition
LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>
chunks
Vec<LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>>