async fn make_chunk(mock_ingester: &mut MockIngester, chunk: ChunkData<'_, '_>) -> String {
    let chunk_stage = chunk.chunk_stage.expect("chunk stage should be set");

    // create chunk
    match chunk_stage {
        ChunkStage::Ingester => {
            // process write
            let (op, _partition_ids) = mock_ingester
                .simulate_write_routing(&chunk.lp_lines, chunk.partition_key)
                .await;
            mock_ingester.buffer_operation(op).await;

            // process delete predicates
            if let Some(delete_table_name) = chunk.delete_table_name {
                for pred in &chunk.preds {
                    match pred.delete_time {
                        DeleteTime::Ingester { .. } => {
                            let op = mock_ingester
                                .simulate_delete_routing(delete_table_name, pred.predicate.clone())
                                .await;
                            mock_ingester.buffer_operation(op).await;
                        }
                        other @ DeleteTime::Parquet => {
                            panic!("Cannot have delete time '{other}' for ingester chunk")
                        }
                        DeleteTime::Begin | DeleteTime::End => {
                            unreachable!(
                                "Begin/end cases should have been replaced \
                                 with concrete instances at this point"
                            )
                        }
                    }
                }
            }
        }
        ChunkStage::Parquet => {
            // process write
            let (op, partition_ids) = mock_ingester
                .simulate_write_routing(&chunk.lp_lines, chunk.partition_key)
                .await;
            mock_ingester.buffer_operation(op).await;

            // model delete predicates that are materialized (applied) by the ingester,
            // during parquet file creation
            let mut tombstone_ids = vec![];
            if let Some(delete_table_name) = chunk.delete_table_name {
                for pred in &chunk.preds {
                    match pred.delete_time {
                        DeleteTime::Ingester { .. } => {
                            let ids_pre = mock_ingester.tombstone_ids(delete_table_name).await;

                            let op = mock_ingester
                                .simulate_delete_routing(delete_table_name, pred.predicate.clone())
                                .await;
                            mock_ingester.buffer_operation(op).await;

                            // tombstones are created immediately, need to remember their ID to
                            // handle deletion later
                            let mut tombstone_id = None;
                            for id in mock_ingester.tombstone_ids(delete_table_name).await {
                                if !ids_pre.contains(&id) {
                                    assert!(tombstone_id.is_none(), "Added multiple tombstones?!");
                                    tombstone_id = Some(id);
                                }
                            }
                            tombstone_ids.push(tombstone_id.expect("No tombstone added?!"));
                        }
                        DeleteTime::Parquet => {
                            // will be attached AFTER the chunk was created
                        }
                        DeleteTime::Begin | DeleteTime::End => {
                            unreachable!(
                                "Begin/end cases should have been replaced \
                                 with concrete instances at this point"
                            )
                        }
                    }
                }
            }

            mock_ingester.persist(&partition_ids).await;

            // model post-persist delete predicates
            if let Some(delete_table_name) = chunk.delete_table_name {
                let mut id_it = tombstone_ids.iter();
                for pred in &chunk.preds {
                    match pred.delete_time {
                        DeleteTime::Ingester { also_in_catalog } => {
                            // should still have a tombstone
                            let tombstone_id = *id_it.next().unwrap();
                            mock_ingester
                                .catalog
                                .catalog
                                .repositories()
                                .await
                                .tombstones()
                                .get_by_id(tombstone_id)
                                .await
                                .unwrap()
                                .expect("tombstone should be present");

                            if !also_in_catalog {
                                mock_ingester
                                    .catalog
                                    .catalog
                                    .repositories()
                                    .await
                                    .tombstones()
                                    .remove(&[tombstone_id])
                                    .await
                                    .unwrap();
                            }
                        }
                        DeleteTime::Parquet => {
                            // create new tombstone
                            let op = mock_ingester
                                .simulate_delete_routing(delete_table_name, pred.predicate.clone())
                                .await;
                            mock_ingester.buffer_operation(op).await;
                        }
                        DeleteTime::Begin | DeleteTime::End => {
                            unreachable!(
                                "Begin/end cases should have been replaced \
                                 with concrete instances at this point"
                            )
                        }
                    }
                }
            }
        }
    }

    let mut name = format!(
        "Chunk stage={} partition={}",
        chunk_stage, chunk.partition_key
    );
    let n_preds = chunk.preds.len();
    if n_preds > 0 {
        let delete_names: Vec<_> = chunk
            .preds
            .iter()
            .map(|p| p.delete_time.to_string())
            .collect();
        write!(
            name,
            " with {} delete(s) ({})",
            n_preds,
            delete_names.join(", ")
        )
        .unwrap();
    }
    name
}
/// Create given chunk using the given ingester.
///
/// Returns a human-readable chunk description.
Create given chunk using the given ingester.
Returns a human-readable chunk description.
Create given chunk using the given ingester . Returns a human - readable chunk description .
async fn make_chunk(mock_ingester: &mut MockIngester, chunk: ChunkData<'_, '_>) -> String {
    let chunk_stage = chunk.chunk_stage.expect("chunk stage should be set");
    // create chunk
    match chunk_stage {
        ChunkStage::Ingester => {
            // process write
            let (op, _partition_ids) = mock_ingester
                .simulate_write_routing(&chunk.lp_lines, chunk.partition_key)
                .await;
            mock_ingester.buffer_operation(op).await;
            // process delete predicates
            if let Some(delete_table_name) = chunk.delete_table_name {
                for pred in &chunk.preds {
                    match pred.delete_time {
                        DeleteTime::Ingester { .. } => {
                            let op = mock_ingester
                                .simulate_delete_routing(delete_table_name, pred.predicate.clone())
                                .await;
                            mock_ingester.buffer_operation(op).await;
                        }
                        other @ DeleteTime::Parquet => {
                            panic!("Cannot have delete time '{other}' for ingester chunk")
                        }
                        DeleteTime::Begin | DeleteTime::End => {
                            unreachable!(
                                "Begin/end cases should have been replaced \
                                 with concrete instances at this point"
                            )
                        }
                    }
                }
            }
        }
        ChunkStage::Parquet => {
            // process write
            let (op, partition_ids) = mock_ingester
                .simulate_write_routing(&chunk.lp_lines, chunk.partition_key)
                .await;
            mock_ingester.buffer_operation(op).await;
            // model delete predicates that are materialized (applied) by the ingester,
            // during parquet file creation
            let mut tombstone_ids = vec![];
            if let Some(delete_table_name) = chunk.delete_table_name {
                for pred in &chunk.preds {
                    match pred.delete_time {
                        DeleteTime::Ingester { .. } => {
                            let ids_pre = mock_ingester.tombstone_ids(delete_table_name).await;
                            let op = mock_ingester
                                .simulate_delete_routing(delete_table_name, pred.predicate.clone())
                                .await;
                            mock_ingester.buffer_operation(op).await;
                            // tombstones are created immediately, need to remember their ID to
                            // handle deletion later
                            let mut tombstone_id = None;
                            for id in mock_ingester.tombstone_ids(delete_table_name).await {
                                if !ids_pre.contains(&id) {
                                    assert!(tombstone_id.is_none(), "Added multiple tombstones?!");
                                    tombstone_id = Some(id);
                                }
                            }
                            tombstone_ids.push(tombstone_id.expect("No tombstone added?!"));
                        }
                        DeleteTime::Parquet => {
                            // will be attached AFTER the chunk was created
                        }
                        DeleteTime::Begin | DeleteTime::End => {
                            unreachable!(
                                "Begin/end cases should have been replaced \
                                 with concrete instances at this point"
                            )
                        }
                    }
                }
            }
            mock_ingester.persist(&partition_ids).await;
            // model post-persist delete predicates
            if let Some(delete_table_name) = chunk.delete_table_name {
                let mut id_it = tombstone_ids.iter();
                for pred in &chunk.preds {
                    match pred.delete_time {
                        DeleteTime::Ingester { also_in_catalog } => {
                            // should still have a tombstone
                            let tombstone_id = *id_it.next().unwrap();
                            mock_ingester
                                .catalog
                                .catalog
                                .repositories()
                                .await
                                .tombstones()
                                .get_by_id(tombstone_id)
                                .await
                                .unwrap()
                                .expect("tombstone should be present");
                            if !also_in_catalog {
                                mock_ingester
                                    .catalog
                                    .catalog
                                    .repositories()
                                    .await
                                    .tombstones()
                                    .remove(&[tombstone_id])
                                    .await
                                    .unwrap();
                            }
                        }
                        DeleteTime::Parquet => {
                            // create new tombstone
                            let op = mock_ingester
                                .simulate_delete_routing(delete_table_name, pred.predicate.clone())
                                .await;
                            mock_ingester.buffer_operation(op).await;
                        }
                        DeleteTime::Begin | DeleteTime::End => {
                            unreachable!(
                                "Begin/end cases should have been replaced \
                                 with concrete instances at this point"
                            )
                        }
                    }
                }
            }
        }
    }
    let mut name = format!(
        "Chunk stage={} partition={}",
        chunk_stage, chunk.partition_key
    );
    let n_preds = chunk.preds.len();
    if n_preds > 0 {
        let delete_names: Vec<_> = chunk
            .preds
            .iter()
            .map(|p| p.delete_time.to_string())
            .collect();
        write!(
            name,
            " with {} delete(s) ({})",
            n_preds,
            delete_names.join(", ")
        )
        .unwrap();
    }
    name
}
async fn make_chunk ( mock_ingester : & mut MockIngester , chunk : ChunkData < ' _ , ' _ > ) -> String { let chunk_stage = chunk . chunk_stage . expect ( " " ) ; // create chunk match chunk_stage { ChunkStage :: Ingester => { // process write let ( op , _partition_ids ) = mock_ingester . simulate_write_routing ( & chunk . lp_lines , chunk . partition_key ) . await ; mock_ingester . buffer_operation ( op ) . await ; // process delete predicates if let Some ( delete_table_name ) = chunk . delete_table_name { for pred in & chunk . preds { match pred . delete_time { DeleteTime :: Ingester { .. } => { let op = mock_ingester . simulate_delete_routing ( delete_table_name , pred . predicate . clone ( ) ) . await ; mock_ingester . buffer_operation ( op ) . await ; } other @ DeleteTime :: Parquet => { panic ! ( " " ) } DeleteTime :: Begin | DeleteTime :: End => { unreachable ! ( " \
 " ) } } } } } ChunkStage :: Parquet => { // process write let ( op , partition_ids ) = mock_ingester . simulate_write_routing ( & chunk . lp_lines , chunk . partition_key ) . await ; mock_ingester . buffer_operation ( op ) . await ; // model delete predicates that are materialized (applied) by the ingester, // during parquet file creation let mut tombstone_ids = vec ! [ ] ; if let Some ( delete_table_name ) = chunk . delete_table_name { for pred in & chunk . preds { match pred . delete_time { DeleteTime :: Ingester { .. } => { let ids_pre = mock_ingester . tombstone_ids ( delete_table_name ) . await ; let op = mock_ingester . simulate_delete_routing ( delete_table_name , pred . predicate . clone ( ) ) . await ; mock_ingester . buffer_operation ( op ) . await ; // tombstones are created immediately, need to remember their ID to // handle deletion later let mut tombstone_id = None ; for id in mock_ingester . tombstone_ids ( delete_table_name ) . await { if ! ids_pre . contains ( & id ) { assert ! ( tombstone_id is_none ( ) " " ) ; tombstone_id = Some ( id ) ; } } tombstone_ids . push ( tombstone_id . expect ( " " ) ) ; } DeleteTime :: Parquet => { // will be attached AFTER the chunk was created } DeleteTime :: Begin | DeleteTime :: End => { unreachable ! ( " \
 " ) } } } } mock_ingester . persist ( & partition_ids ) . await ; // model post-persist delete predicates if let Some ( delete_table_name ) = chunk . delete_table_name { let mut id_it = tombstone_ids . iter ( ) ; for pred in & chunk . preds { match pred . delete_time { DeleteTime :: Ingester { also_in_catalog } => { // should still have a tombstone let tombstone_id = * id_it . next ( ) . unwrap ( ) ; mock_ingester . catalog . catalog . repositories ( ) . await . tombstones ( ) . get_by_id ( tombstone_id ) . await . unwrap ( ) . expect ( " " ) ; if ! also_in_catalog { mock_ingester . catalog . catalog . repositories ( ) . await . tombstones ( ) . remove ( & [ tombstone_id ] ) . await . unwrap ( ) ; } } DeleteTime :: Parquet => { // create new tombstone let op = mock_ingester . simulate_delete_routing ( delete_table_name , pred . predicate . clone ( ) ) . await ; mock_ingester . buffer_operation ( op ) . await ; } DeleteTime :: Begin | DeleteTime :: End => { unreachable ! ( " \
 " ) } } } } } } let mut name = format ! ( " " chunk_stage chunk partition_key ) ; let n_preds = chunk . preds . len ( ) ; if n_preds > 0 { let delete_names : Vec < _ > = chunk . preds . iter ( ) . map ( | p | p . delete_time . to_string ( ) ) . collect ( ) ; write ! ( name " " n_preds delete_names join ( " " ) ) . unwrap ( ) ; } name }
Create given chunk using the given ingester.
Create given chunk using the given ingester .
mock_ingester
&mut MockIngester
chunk
ChunkData<'_, '_>
mock_ingester
&mut MockIngester
chunk
ChunkData<'_, '_>