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

    info!(%table_name, %partition_key, ?chunk_ids, "splitting and persisting chunks");

    let max_persistable_timestamp = flush_handle.timestamp();
    let flush_timestamp = max_persistable_timestamp.timestamp_nanos();

    let (tracker, registration) = db.jobs.register(Job::PersistChunks {
        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 mut query_chunks = vec![];
    for mut chunk in chunks {
        // 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_writing_to_object_store(&registration)?;
        query_chunks.push(DbChunk::snapshot(&*chunk));
    }

    // drop partition lock guard
    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 key = compute_sort_key(query_chunks.iter().map(|x| x.summary()));
        let key_str = format!("\"{}\"", key); // for logging

        // build schema
        let schema = merge_schemas(&query_chunks);

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

        let physical_plan = ctx.prepare_plan(&plan)?;
        assert_eq!(
            physical_plan.output_partitioning().partition_count(),
            2,
            "Expected split plan to produce exactly 2 partitions"
        );

        let to_persist_stream = ctx.execute_partition(Arc::clone(&physical_plan), 0).await?;
        let remainder_stream = ctx.execute_partition(physical_plan, 1).await?;

        let (to_persist, remainder) = futures::future::try_join(
            collect_rub(to_persist_stream, db.as_ref(), &table_name),
            collect_rub(remainder_stream, db.as_ref(), &table_name),
        )
        .await?;

        let persisted_rows = to_persist.as_ref().map(|p| p.rows()).unwrap_or(0);
        let remainder_rows = remainder.as_ref().map(|r| r.rows()).unwrap_or(0);

        let persist_fut = {
            let partition = LockableCatalogPartition::new(Arc::clone(&db), partition);
            let mut partition_write = partition.write();
            for id in chunk_ids {
                partition_write.force_drop_chunk(id)
            }

            // Upsert remainder to catalog
            if let Some(remainder) = remainder {
                partition_write.create_rub_chunk(
                    remainder,
                    time_of_first_write,
                    time_of_last_write,
                    Arc::clone(&schema),
                );
            }

            let to_persist = to_persist.expect("should be rows to persist");

            let to_persist = LockableCatalogChunk {
                db,
                chunk: partition_write.create_rub_chunk(
                    to_persist,
                    time_of_first_write,
                    time_of_last_write,
                    schema,
                ),
            };
            let to_persist = to_persist.write();

            write_chunk_to_object_store(partition_write, to_persist, flush_handle)?.1
        };

        // Wait for write operation to complete
        let persisted_chunk = persist_fut.await??;

        let elapsed = now.elapsed();
        // input rows per second
        let throughput = (input_rows as u128 * 1_000_000_000) / elapsed.as_nanos();

        info!(input_chunks=query_chunks.len(),
              input_rows, persisted_rows, remainder_rows,
              sort_key=%key_str, compaction_took = ?elapsed,
              ?max_persistable_timestamp,
              rows_per_sec=?throughput,  "chunk(s) persisted");

        Ok(persisted_chunk)
    };

    Ok((tracker, fut.track(registration)))
}
/// Split and then persist the provided chunks
///
/// TODO: Replace low-level locks with transaction object
Split and then persist the provided chunks
TODO: Replace low-level locks with transaction object
Split and then persist the provided chunks TODO : Replace low - level locks with transaction object
pub fn persist_chunks(
    partition: LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>,
    chunks: Vec<LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>>,
    flush_handle: FlushHandle,
) -> Result<(
    TaskTracker<Job>,
    TrackedFuture<impl Future<Output = Result<Arc<DbChunk>>> + Send>,
)> {
    let now = std::time::Instant::now(); // time persist duration.
    let db = Arc::clone(&partition.data().db);
    let table_name = partition.table_name().to_string();
    let partition_key = partition.key().to_string();
    let chunk_ids: Vec<_> = chunks.iter().map(|x| x.id()).collect();
    info!(%table_name, %partition_key, ?chunk_ids, "splitting and persisting chunks");
    let max_persistable_timestamp = flush_handle.timestamp();
    let flush_timestamp = max_persistable_timestamp.timestamp_nanos();
    let (tracker, registration) = db.jobs.register(Job::PersistChunks {
        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 mut query_chunks = vec![];
    for mut chunk in chunks {
        // 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_writing_to_object_store(&registration)?;
        query_chunks.push(DbChunk::snapshot(&*chunk));
    }
    // drop partition lock guard
    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 key = compute_sort_key(query_chunks.iter().map(|x| x.summary()));
        let key_str = format!("\"{}\"", key); // for logging
        // build schema
        let schema = merge_schemas(&query_chunks);
        // Cannot move query_chunks as the sort key borrows the column names
        let (schema, plan) = ReorgPlanner::new().split_plan(
            schema,
            query_chunks.iter().map(Arc::clone),
            key,
            flush_timestamp,
        )?;
        let physical_plan = ctx.prepare_plan(&plan)?;
        assert_eq!(
            physical_plan.output_partitioning().partition_count(),
            2,
            "Expected split plan to produce exactly 2 partitions"
        );
        let to_persist_stream = ctx.execute_partition(Arc::clone(&physical_plan), 0).await?;
        let remainder_stream = ctx.execute_partition(physical_plan, 1).await?;
        let (to_persist, remainder) = futures::future::try_join(
            collect_rub(to_persist_stream, db.as_ref(), &table_name),
            collect_rub(remainder_stream, db.as_ref(), &table_name),
        )
        .await?;
        let persisted_rows = to_persist.as_ref().map(|p| p.rows()).unwrap_or(0);
        let remainder_rows = remainder.as_ref().map(|r| r.rows()).unwrap_or(0);
        let persist_fut = {
            let partition = LockableCatalogPartition::new(Arc::clone(&db), partition);
            let mut partition_write = partition.write();
            for id in chunk_ids {
                partition_write.force_drop_chunk(id)
            }
            // Upsert remainder to catalog
            if let Some(remainder) = remainder {
                partition_write.create_rub_chunk(
                    remainder,
                    time_of_first_write,
                    time_of_last_write,
                    Arc::clone(&schema),
                );
            }
            let to_persist = to_persist.expect("should be rows to persist");
            let to_persist = LockableCatalogChunk {
                db,
                chunk: partition_write.create_rub_chunk(
                    to_persist,
                    time_of_first_write,
                    time_of_last_write,
                    schema,
                ),
            };
            let to_persist = to_persist.write();
            write_chunk_to_object_store(partition_write, to_persist, flush_handle)?.1
        };
        // Wait for write operation to complete
        let persisted_chunk = persist_fut.await??;
        let elapsed = now.elapsed();
        // input rows per second
        let throughput = (input_rows as u128 * 1_000_000_000) / elapsed.as_nanos();
        info!(input_chunks=query_chunks.len(),
              input_rows, persisted_rows, remainder_rows,
              sort_key=%key_str, compaction_took = ?elapsed,
              ?max_persistable_timestamp,
              rows_per_sec=?throughput,  "chunk(s) persisted");
        Ok(persisted_chunk)
    };
    Ok((tracker, fut.track(registration)))
}
pub fn persist_chunks ( partition : LifecycleWriteGuard < ' _ , Partition , LockableCatalogPartition > , chunks : Vec < LifecycleWriteGuard < ' _ , CatalogChunk , LockableCatalogChunk > > , flush_handle : FlushHandle , ) -> Result < ( TaskTracker < Job > , TrackedFuture < impl Future < Output = Result < Arc < DbChunk > > > + Send > , ) > { let now = std :: time :: Instant :: now ( ) ; // time persist duration. let db = Arc :: clone ( & partition . data ( ) . db ) ; let table_name = partition . table_name ( ) . to_string ( ) ; let partition_key = partition . key ( ) . to_string ( ) ; let chunk_ids : Vec < _ > = chunks . iter ( ) . map ( | x | x . id ( ) ) . collect ( ) ; info ! ( table_name partition_key chunk_ids " " ) ; let max_persistable_timestamp = flush_handle . timestamp ( ) ; let flush_timestamp = max_persistable_timestamp . timestamp_nanos ( ) ; let ( tracker , registration ) = db . jobs . register ( Job :: PersistChunks { 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 mut query_chunks = vec ! [ ] ; for mut chunk in chunks { // 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_writing_to_object_store ( & registration ) ? ; query_chunks . push ( DbChunk :: snapshot ( & * chunk ) ) ; } // drop partition lock guard 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 key = compute_sort_key ( query_chunks . iter ( ) . map ( | x | x . summary ( ) ) ) ; let key_str = format ! ( " \" \" " key ) ; // for logging // build schema let schema = merge_schemas ( & query_chunks ) ; // Cannot move query_chunks as the sort key borrows the column names let ( schema , plan ) = ReorgPlanner :: new ( ) . split_plan ( schema , query_chunks . iter ( ) . map ( Arc :: clone ) , key , flush_timestamp , ) ? ; let physical_plan = ctx . prepare_plan ( & plan ) ? ; assert_eq ! ( physical_plan output_partitioning ( ) partition_count ( ) 2 " " ) ; let to_persist_stream = ctx . execute_partition ( Arc :: clone ( & physical_plan ) , 0 ) . await ? ; let remainder_stream = ctx . execute_partition ( physical_plan , 1 ) . await ? ; let ( to_persist , remainder ) = futures :: future :: try_join ( collect_rub ( to_persist_stream , db . as_ref ( ) , & table_name ) , collect_rub ( remainder_stream , db . as_ref ( ) , & table_name ) , ) . await ? ; let persisted_rows = to_persist . as_ref ( ) . map ( | p | p . rows ( ) ) . unwrap_or ( 0 ) ; let remainder_rows = remainder . as_ref ( ) . map ( | r | r . rows ( ) ) . unwrap_or ( 0 ) ; let persist_fut = { let partition = LockableCatalogPartition :: new ( Arc :: clone ( & db ) , partition ) ; let mut partition_write = partition . write ( ) ; for id in chunk_ids { partition_write . force_drop_chunk ( id ) } // Upsert remainder to catalog if let Some ( remainder ) = remainder { partition_write . create_rub_chunk ( remainder , time_of_first_write , time_of_last_write , Arc :: clone ( & schema ) , ) ; } let to_persist = to_persist . expect ( " " ) ; let to_persist = LockableCatalogChunk { db , chunk : partition_write . create_rub_chunk ( to_persist , time_of_first_write , time_of_last_write , schema , ) , } ; let to_persist = to_persist . write ( ) ; write_chunk_to_object_store ( partition_write , to_persist , flush_handle ) ? . 1 } ; // Wait for write operation to complete let persisted_chunk = persist_fut . await ? ? ; let elapsed = now . elapsed ( ) ; // input rows per second let throughput = ( input_rows as u128 * 1_000_000_000 ) / elapsed . as_nanos ( ) ; info ! ( input_chunks query_chunks len ( ) input_rows persisted_rows remainder_rows sort_key key_str compaction_took elapsed max_persistable_timestamp rows_per_sec throughput " " ) ; Ok ( persisted_chunk ) } ; Ok ( ( tracker , fut . track ( registration ) ) ) }
Split and then persist the provided chunks
TODO: Replace low-level locks with transaction object
Split and then persist the provided chunks TODO : Replace low - level locks with transaction object
partition
LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>
chunks
Vec<LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>>
flush_handle
FlushHandle
partition
LifecycleWriteGuard<'_, Partition, LockableCatalogPartition>
chunks
Vec<LifecycleWriteGuard<'_, CatalogChunk, LockableCatalogChunk>>
flush_handle
FlushHandle