async fn compact(
        &self,
        overlapped_files: Vec<ParquetFileWithTombstone>,
        partition: &Partition,
    ) -> Result<Vec<CompactedData>> {
        debug!(num_files = overlapped_files.len(), "compact files");

        // Nothing to compact
        if overlapped_files.is_empty() {
            return Ok(vec![]);
        }

        // One file without tombstone, no need to compact
        if overlapped_files.len() == 1 && overlapped_files[0].tombstones().is_empty() {
            return Ok(vec![]);
        }

        // Save the parquet metadata for the first file to reuse IDs and names
        let iox_metadata = overlapped_files[0].iox_metadata();

        // Hold onto various metadata items needed later.
        let sequencer_id = overlapped_files[0].sequencer_id;
        let namespace_id = overlapped_files[0].namespace_id;
        let table_id = overlapped_files[0].table_id;

        //  Collect all unique tombstone
        let mut tombstone_map = overlapped_files[0].tombstone_map();

        // Verify if the given files belong to the same partition and collect their tombstones
        //  One tombstone might be relevant to multiple parquet files in this set, so dedupe here.
        if let Some((head, tail)) = overlapped_files.split_first() {
            for file in tail {
                tombstone_map.append(&mut file.tombstone_map());

                let is_same = file.sequencer_id == head.sequencer_id
                    && file.table_id == head.table_id
                    && file.partition_id == head.partition_id;

                ensure!(
                    is_same,
                    NotSamePartitionSnafu {
                        sequencer_id_1: head.sequencer_id,
                        table_id_1: head.table_id,
                        partition_id_1: head.partition_id,
                        sequencer_id_2: file.sequencer_id,
                        table_id_2: file.table_id,
                        partition_id_2: file.partition_id
                    }
                );

                assert_eq!(file.sequencer_id, sequencer_id);
                assert_eq!(file.namespace_id, namespace_id);
                assert_eq!(file.table_id, table_id);
                assert_eq!(file.partition_id, partition.id);
            }
        }

        // Convert the input files into QueryableParquetChunk for making query
        // plan
        let query_chunks: Vec<_> = overlapped_files
            .iter()
            .map(|f| {
                f.to_queryable_parquet_chunk(
                    self.store.clone(),
                    iox_metadata.table_name.to_string(),
                    f.iox_metadata().sort_key,
                    partition.sort_key(),
                )
            })
            .collect();

        trace!(
            n_query_chunks = query_chunks.len(),
            "gathered parquet data to compact"
        );

        // Compute min & max sequence numbers and time
        // unwrap here will work becasue the len of the query_chunks already >= 1
        let (head, tail) = query_chunks.split_first().unwrap();
        let mut min_sequence_number = head.min_sequence_number();
        let mut max_sequence_number = head.max_sequence_number();
        let mut min_time = head.min_time();
        let mut max_time = head.max_time();
        for c in tail {
            min_sequence_number = min(min_sequence_number, c.min_sequence_number());
            max_sequence_number = max(max_sequence_number, c.max_sequence_number());
            min_time = min(min_time, c.min_time());
            max_time = max(max_time, c.max_time());
        }

        // Merge schema of the compacting chunks
        let query_chunks: Vec<_> = query_chunks
            .into_iter()
            .map(|c| Arc::new(c) as Arc<dyn QueryChunk>)
            .collect();
        let merged_schema = QueryableParquetChunk::merge_schemas(&query_chunks);
        debug!(
            num_cols = merged_schema.as_arrow().fields().len(),
            "Number of columns in the merged schema to build query plan"
        );

        // All partitions in the catalog MUST contain a sort key.
        let sort_key = partition
            .sort_key()
            .expect("no parittion sort key in catalog")
            .filter_to(&merged_schema.primary_key());

        // Identify split time
        let split_time = self.compute_split_time(min_time, max_time);

        // Build compact logical plan
        let plan = {
            // split data to compact data into 2 files
            if min_time < split_time && split_time < max_time {
                // split compact query plan
                ReorgPlanner::new()
                    .split_plan(
                        Arc::clone(&merged_schema),
                        query_chunks,
                        sort_key.clone(),
                        split_time,
                    )
                    .context(CompactLogicalPlanSnafu)?
            } else {
                // compact everything into one file
                ReorgPlanner::new()
                    .compact_plan(Arc::clone(&merged_schema), query_chunks, sort_key.clone())
                    .context(CompactLogicalPlanSnafu)?
            }
        };

        let ctx = self.exec.new_context(ExecutorType::Reorg);
        let physical_plan = ctx
            .prepare_plan(&plan)
            .await
            .context(CompactPhysicalPlanSnafu)?;

        // Run to collect each stream of the plan
        let stream_count = physical_plan.output_partitioning().partition_count();
        let mut compacted = Vec::with_capacity(stream_count);
        debug!("running plan with {} streams", stream_count);
        for i in 0..stream_count {
            trace!(partition = i, "executing datafusion partition");

            let stream = ctx
                .execute_stream_partitioned(Arc::clone(&physical_plan), i)
                .await
                .context(ExecuteCompactPlanSnafu)?;

            trace!(partition = i, "built result stream for partition");

            let meta = IoxMetadata {
                object_store_id: Uuid::new_v4(),
                creation_timestamp: self.time_provider.now(),
                sequencer_id: iox_metadata.sequencer_id,
                namespace_id: iox_metadata.namespace_id,
                namespace_name: Arc::<str>::clone(&iox_metadata.namespace_name),
                table_id: iox_metadata.table_id,
                table_name: Arc::<str>::clone(&iox_metadata.table_name),
                partition_id: iox_metadata.partition_id,
                partition_key: Arc::<str>::clone(&iox_metadata.partition_key),
                min_sequence_number,
                max_sequence_number,
                compaction_level: 1, // compacted result file always have level 1
                sort_key: Some(sort_key.clone()),
            };

            let compacted_data = CompactedData::new(stream, meta, tombstone_map.clone());
            compacted.push(compacted_data);
        }

        Ok(compacted)
    }
// Compact given files. The given files are either overlapped or contiguous in time.
// The output will include 2 CompactedData sets, one contains a large amount of data of
// least recent time and the other has a small amount of data of most recent time. Each
// will be persisted in its own file. The idea is when new writes come, they will
// mostly overlap with the most recent data only.
Compact given files. The given files are either overlapped or contiguous in time.
The output will include 2 CompactedData sets, one contains a large amount of data of
least recent time and the other has a small amount of data of most recent time. Each
will be persisted in its own file. The idea is when new writes come, they will
mostly overlap with the most recent data only.
Compact given files . The given files are either overlapped or contiguous in time . The output will include 2 CompactedData sets one contains a large amount of data of least recent time and the other has a small amount of data of most recent time . Each will be persisted in its own file . The idea is when new writes come they will mostly overlap with the most recent data only .
async fn compact(
        &self,
        overlapped_files: Vec<ParquetFileWithTombstone>,
        partition: &Partition,
    ) -> Result<Vec<CompactedData>> {
        debug!(num_files = overlapped_files.len(), "compact files");
        // Nothing to compact
        if overlapped_files.is_empty() {
            return Ok(vec![]);
        }
        // One file without tombstone, no need to compact
        if overlapped_files.len() == 1 && overlapped_files[0].tombstones().is_empty() {
            return Ok(vec![]);
        }
        // Save the parquet metadata for the first file to reuse IDs and names
        let iox_metadata = overlapped_files[0].iox_metadata();
        // Hold onto various metadata items needed later.
        let sequencer_id = overlapped_files[0].sequencer_id;
        let namespace_id = overlapped_files[0].namespace_id;
        let table_id = overlapped_files[0].table_id;
        //  Collect all unique tombstone
        let mut tombstone_map = overlapped_files[0].tombstone_map();
        // Verify if the given files belong to the same partition and collect their tombstones
        //  One tombstone might be relevant to multiple parquet files in this set, so dedupe here.
        if let Some((head, tail)) = overlapped_files.split_first() {
            for file in tail {
                tombstone_map.append(&mut file.tombstone_map());
                let is_same = file.sequencer_id == head.sequencer_id
                    && file.table_id == head.table_id
                    && file.partition_id == head.partition_id;
                ensure!(
                    is_same,
                    NotSamePartitionSnafu {
                        sequencer_id_1: head.sequencer_id,
                        table_id_1: head.table_id,
                        partition_id_1: head.partition_id,
                        sequencer_id_2: file.sequencer_id,
                        table_id_2: file.table_id,
                        partition_id_2: file.partition_id
                    }
                );
                assert_eq!(file.sequencer_id, sequencer_id);
                assert_eq!(file.namespace_id, namespace_id);
                assert_eq!(file.table_id, table_id);
                assert_eq!(file.partition_id, partition.id);
            }
        }
        // Convert the input files into QueryableParquetChunk for making query
        // plan
        let query_chunks: Vec<_> = overlapped_files
            .iter()
            .map(|f| {
                f.to_queryable_parquet_chunk(
                    self.store.clone(),
                    iox_metadata.table_name.to_string(),
                    f.iox_metadata().sort_key,
                    partition.sort_key(),
                )
            })
            .collect();
        trace!(
            n_query_chunks = query_chunks.len(),
            "gathered parquet data to compact"
        );
        // Compute min & max sequence numbers and time
        // unwrap here will work becasue the len of the query_chunks already >= 1
        let (head, tail) = query_chunks.split_first().unwrap();
        let mut min_sequence_number = head.min_sequence_number();
        let mut max_sequence_number = head.max_sequence_number();
        let mut min_time = head.min_time();
        let mut max_time = head.max_time();
        for c in tail {
            min_sequence_number = min(min_sequence_number, c.min_sequence_number());
            max_sequence_number = max(max_sequence_number, c.max_sequence_number());
            min_time = min(min_time, c.min_time());
            max_time = max(max_time, c.max_time());
        }
        // Merge schema of the compacting chunks
        let query_chunks: Vec<_> = query_chunks
            .into_iter()
            .map(|c| Arc::new(c) as Arc<dyn QueryChunk>)
            .collect();
        let merged_schema = QueryableParquetChunk::merge_schemas(&query_chunks);
        debug!(
            num_cols = merged_schema.as_arrow().fields().len(),
            "Number of columns in the merged schema to build query plan"
        );
        // All partitions in the catalog MUST contain a sort key.
        let sort_key = partition
            .sort_key()
            .expect("no parittion sort key in catalog")
            .filter_to(&merged_schema.primary_key());
        // Identify split time
        let split_time = self.compute_split_time(min_time, max_time);
        // Build compact logical plan
        let plan = {
            // split data to compact data into 2 files
            if min_time < split_time && split_time < max_time {
                // split compact query plan
                ReorgPlanner::new()
                    .split_plan(
                        Arc::clone(&merged_schema),
                        query_chunks,
                        sort_key.clone(),
                        split_time,
                    )
                    .context(CompactLogicalPlanSnafu)?
            } else {
                // compact everything into one file
                ReorgPlanner::new()
                    .compact_plan(Arc::clone(&merged_schema), query_chunks, sort_key.clone())
                    .context(CompactLogicalPlanSnafu)?
            }
        };
        let ctx = self.exec.new_context(ExecutorType::Reorg);
        let physical_plan = ctx
            .prepare_plan(&plan)
            .await
            .context(CompactPhysicalPlanSnafu)?;
        // Run to collect each stream of the plan
        let stream_count = physical_plan.output_partitioning().partition_count();
        let mut compacted = Vec::with_capacity(stream_count);
        debug!("running plan with {} streams", stream_count);
        for i in 0..stream_count {
            trace!(partition = i, "executing datafusion partition");
            let stream = ctx
                .execute_stream_partitioned(Arc::clone(&physical_plan), i)
                .await
                .context(ExecuteCompactPlanSnafu)?;
            trace!(partition = i, "built result stream for partition");
            let meta = IoxMetadata {
                object_store_id: Uuid::new_v4(),
                creation_timestamp: self.time_provider.now(),
                sequencer_id: iox_metadata.sequencer_id,
                namespace_id: iox_metadata.namespace_id,
                namespace_name: Arc::<str>::clone(&iox_metadata.namespace_name),
                table_id: iox_metadata.table_id,
                table_name: Arc::<str>::clone(&iox_metadata.table_name),
                partition_id: iox_metadata.partition_id,
                partition_key: Arc::<str>::clone(&iox_metadata.partition_key),
                min_sequence_number,
                max_sequence_number,
                compaction_level: 1, // compacted result file always have level 1
                sort_key: Some(sort_key.clone()),
            };
            let compacted_data = CompactedData::new(stream, meta, tombstone_map.clone());
            compacted.push(compacted_data);
        }
        Ok(compacted)
    }
async fn compact ( & self , overlapped_files : Vec < ParquetFileWithTombstone > , partition : & Partition , ) -> Result < Vec < CompactedData > > { debug ! ( num_files overlapped_files len ( ) " " ) ; // Nothing to compact if overlapped_files . is_empty ( ) { return Ok ( vec ! [ ] ) ; } // One file without tombstone, no need to compact if overlapped_files . len ( ) == 1 && overlapped_files [ 0 ] . tombstones ( ) . is_empty ( ) { return Ok ( vec ! [ ] ) ; } // Save the parquet metadata for the first file to reuse IDs and names let iox_metadata = overlapped_files [ 0 ] . iox_metadata ( ) ; // Hold onto various metadata items needed later. let sequencer_id = overlapped_files [ 0 ] . sequencer_id ; let namespace_id = overlapped_files [ 0 ] . namespace_id ; let table_id = overlapped_files [ 0 ] . table_id ; //  Collect all unique tombstone let mut tombstone_map = overlapped_files [ 0 ] . tombstone_map ( ) ; // Verify if the given files belong to the same partition and collect their tombstones //  One tombstone might be relevant to multiple parquet files in this set, so dedupe here. if let Some ( ( head , tail ) ) = overlapped_files . split_first ( ) { for file in tail { tombstone_map . append ( & mut file . tombstone_map ( ) ) ; let is_same = file . sequencer_id == head . sequencer_id && file . table_id == head . table_id && file . partition_id == head . partition_id ; ensure ! ( is_same NotSamePartitionSnafu { sequencer_id_1 head sequencer_id table_id_1 head table_id partition_id_1 head partition_id sequencer_id_2 file sequencer_id table_id_2 file table_id partition_id_2 file partition_id } ) ; assert_eq ! ( file sequencer_id sequencer_id ) ; assert_eq ! ( file namespace_id namespace_id ) ; assert_eq ! ( file table_id table_id ) ; assert_eq ! ( file partition_id partition id ) ; } } // Convert the input files into QueryableParquetChunk for making query // plan let query_chunks : Vec < _ > = overlapped_files . iter ( ) . map ( | f | { f . to_queryable_parquet_chunk ( self . store . clone ( ) , iox_metadata . table_name . to_string ( ) , f . iox_metadata ( ) . sort_key , partition . sort_key ( ) , ) } ) . collect ( ) ; trace ! ( n_query_chunks query_chunks len ( ) " " ) ; // Compute min & max sequence numbers and time // unwrap here will work becasue the len of the query_chunks already >= 1 let ( head , tail ) = query_chunks . split_first ( ) . unwrap ( ) ; let mut min_sequence_number = head . min_sequence_number ( ) ; let mut max_sequence_number = head . max_sequence_number ( ) ; let mut min_time = head . min_time ( ) ; let mut max_time = head . max_time ( ) ; for c in tail { min_sequence_number = min ( min_sequence_number , c . min_sequence_number ( ) ) ; max_sequence_number = max ( max_sequence_number , c . max_sequence_number ( ) ) ; min_time = min ( min_time , c . min_time ( ) ) ; max_time = max ( max_time , c . max_time ( ) ) ; } // Merge schema of the compacting chunks let query_chunks : Vec < _ > = query_chunks . into_iter ( ) . map ( | c | Arc :: new ( c ) as Arc < dyn QueryChunk > ) . collect ( ) ; let merged_schema = QueryableParquetChunk :: merge_schemas ( & query_chunks ) ; debug ! ( num_cols merged_schema as_arrow ( ) fields ( ) len ( ) " " ) ; // All partitions in the catalog MUST contain a sort key. let sort_key = partition . sort_key ( ) . expect ( " " ) . filter_to ( & merged_schema . primary_key ( ) ) ; // Identify split time let split_time = self . compute_split_time ( min_time , max_time ) ; // Build compact logical plan let plan = { // split data to compact data into 2 files if min_time < split_time && split_time < max_time { // split compact query plan ReorgPlanner :: new ( ) . split_plan ( Arc :: clone ( & merged_schema ) , query_chunks , sort_key . clone ( ) , split_time , ) . context ( CompactLogicalPlanSnafu ) ? } else { // compact everything into one file ReorgPlanner :: new ( ) . compact_plan ( Arc :: clone ( & merged_schema ) , query_chunks , sort_key . clone ( ) ) . context ( CompactLogicalPlanSnafu ) ? } } ; let ctx = self . exec . new_context ( ExecutorType :: Reorg ) ; let physical_plan = ctx . prepare_plan ( & plan ) . await . context ( CompactPhysicalPlanSnafu ) ? ; // Run to collect each stream of the plan let stream_count = physical_plan . output_partitioning ( ) . partition_count ( ) ; let mut compacted = Vec :: with_capacity ( stream_count ) ; debug ! ( " " stream_count ) ; for i in 0 .. stream_count { trace ! ( partition i " " ) ; let stream = ctx . execute_stream_partitioned ( Arc :: clone ( & physical_plan ) , i ) . await . context ( ExecuteCompactPlanSnafu ) ? ; trace ! ( partition i " " ) ; let meta = IoxMetadata { object_store_id : Uuid :: new_v4 ( ) , creation_timestamp : self . time_provider . now ( ) , sequencer_id : iox_metadata . sequencer_id , namespace_id : iox_metadata . namespace_id , namespace_name : Arc :: < str > :: clone ( & iox_metadata . namespace_name ) , table_id : iox_metadata . table_id , table_name : Arc :: < str > :: clone ( & iox_metadata . table_name ) , partition_id : iox_metadata . partition_id , partition_key : Arc :: < str > :: clone ( & iox_metadata . partition_key ) , min_sequence_number , max_sequence_number , compaction_level : 1 , // compacted result file always have level 1 sort_key : Some ( sort_key . clone ( ) ) , } ; let compacted_data = CompactedData :: new ( stream , meta , tombstone_map . clone ( ) ) ; compacted . push ( compacted_data ) ; } Ok ( compacted ) }
Compact given files.
Compact given files .
&self
overlapped_files
Vec<ParquetFileWithTombstone>
partition
&Partition
&self
overlapped_files
Vec<ParquetFileWithTombstone>
partition
&Partition