时序数据库Influx-IOx源码学习八（Chunk持久化）

ChunkState::Moved(_) if would_write => {
                    let partition_key = chunk_guard.key().to_string();
                    let table_name = chunk_guard.table_name().to_string();
                    let chunk_id = chunk_guard.id();
                    std::mem::drop(chunk_guard);
                    write_active = true;
                    //处于Moved状态下的Chunk会调用write_to_object_store方法进行持久化
                    self.write_to_object_store(partition_key, table_name, chunk_id);
                }

//write_to_object_store实际调用到write_chunk_to_object_store_in_background方法来进行持久化
pub fn write_chunk_to_object_store_in_background(
        self: &Arc<Self>,
        partition_key: String,
        table_name: String,
        chunk_id: u32,
    ) -> TaskTracker<Job> {
        //获取数据库名称
        let name = self.rules.read().name.clone();
        //新建一个后台任务的管理器，用来记录db中都在执行哪些任务及状态，
        let (tracker, registration) = self.jobs.register(Job::WriteChunk {
            db_name: name.to_string(),
            partition_key: partition_key.clone(),
            table_name: table_name.clone(),
            chunk_id,
        });

        let captured = Arc::clone(&self);
        //异步写入
        let task = async move {
            let result = captured
                //真正的写入方法
                .write_chunk_to_object_store(&partition_key, &table_name, chunk_id)
                .await;
            if let Err(e) = result {
                info!(?e, %name, %partition_key, %chunk_id, "background task error loading object store chunk");
                return Err(e);
            }
            Ok(())
        };

        tokio::spawn(task.track(registration));

        tracker
    }

  pub async fn write_chunk_to_object_store(
        &self,
        partition_key: &str,
        table_name: &str,
        chunk_id: u32,
    ) -> Result<Arc<DbChunk>> {
        //从catalog中取回chunk
        let chunk = {
            //先找partition
            let partition =
                self.catalog
                    .valid_partition(partition_key)
                    .context(LoadingChunkToParquet {
                        partition_key,
                        table_name,
                        chunk_id,
                    })?;
            let partition = partition.read();
            //从partition里根据表名和chunk_id拿到chunk
            partition
                .chunk(table_name, chunk_id)
                .context(LoadingChunkToParquet {
                    partition_key,
                    table_name,
                    chunk_id,
                })?
        };


        let rb_chunk = {
            //先加写锁
            let mut chunk = chunk.write();
            //修改Chunk的状态为WritingToObjectStore
            chunk
                .set_writing_to_object_store()
                .context(LoadingChunkToParquet {
                    partition_key,
                    table_name,
                    chunk_id,
                })?
        };

        //获取所有Chunk下所有表的Statistics信息
        let table_stats = rb_chunk.table_summaries();

        //创建一个parquet Chunk，这个在上一章里有提到各种Chunk类型
        let mut parquet_chunk = Chunk::new(
            partition_key.to_string(),
            chunk_id,
            //用来统计parquet占用的内存
            self.memory_registries.parquet.as_ref(),
        );
        //创建一个Storage结构，使用的是启动数据库时候指定的存储类型，这个在第3章里有提到
        let storage = Storage::new(
            Arc::clone(&self.store),
            self.server_id,
            self.rules.read().name.to_string(),
        );
        //遍历所有表的统计数据
        for stats in table_stats {
            //构建一个空的查询，也就是 select * from table，不加where
            let predicate = read_buffer::Predicate::default();

            //从rb_chunk筛选数据， Selection::All代表所有列，predicate代表没有where条件
            //意思就是 `stats` 指向的单个表内的所有数据
            let read_results = rb_chunk
                .read_filter(stats.name.as_str(), predicate, Selection::All)
                .context(ReadBufferChunkError {
                    table_name,
                    chunk_id,
                })?;
            //再拿出来schema信息，因为arrow是分开存的，所以需要拿两次
            let arrow_schema: ArrowSchemaRef = rb_chunk
                .read_filter_table_schema(stats.name.as_str(), Selection::All)
                .context(ReadBufferChunkSchemaError {
                    table_name,
                    chunk_id,
                })?
                .into();
            //再拿出来这个表里的最大最小的时间
            //这个是从readBuffer::Column::from里完成的最大最小时间统计
            //也就是当从mutbuffer转移到readbuffer的时候
            let time_range = rb_chunk.table_time_range(stats.name.as_str()).context(
                ReadBufferChunkTimestampError {
                    table_name,
                    chunk_id,
                },
            )?;
            //创建一个ReadFilterResultsStream
            //官方文档里面说的是这是一个转变ReadFilterResults为异步流的适配器
            let stream: SendableRecordBatchStream = Box::pin(
                streams::ReadFilterResultsStream::new(read_results, Arc::clone(&arrow_schema)),
            );

            // 写到持久化存储当中
            let path = storage
                .write_to_object_store(
                    partition_key.to_string(),
                    chunk_id,
                    stats.name.to_string(),
                    stream,
                )
                .await
                .context(WritingToObjectStore)?;

            // 这里就是把写入parquet的摘要信息存储在内存中
            let schema = Arc::clone(&arrow_schema)
                .try_into()
                .context(SchemaConversion)?;
            let table_time_range = time_range.map(|(start, end)| TimestampRange::new(start, end));
            parquet_chunk.add_table(stats, path, schema, table_time_range);
        }

        //对`catlog::chunk`加写锁，然后更新这个chunk的状态为WrittenToObjectStore
        let mut chunk = chunk.write();
        let parquet_chunk = Arc::clone(&Arc::new(parquet_chunk));
        chunk
            .set_written_to_object_store(parquet_chunk)
            .context(LoadingChunkToParquet {
                partition_key,
                table_name,
                chunk_id,
            })?;
        //包装`catlog::chunk`为`ParquetChunk`
        Ok(DbChunk::snapshot(&chunk))
    }

pub enum ChunkState {
    ....省略
    //这里就是mutbuffer里的chunk
    Moving(Arc<MBChunk>),
    //这里就变成存储的readbuffer的chunk结构
    Moved(Arc<ReadBufferChunk>),
    //这里又开始存储ParquetChunk结构
    WrittenToObjectStore(Arc<ReadBufferChunk>, Arc<ParquetChunk>),
}

    //这里直接跳跃到ObjectStore的put方法里，来看怎么组织的写入
    async fn put<S>(&self, location: &Self::Path, bytes: S, length: Option<usize>) -> Result<()>
    where
        S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
    {
        use ObjectStoreIntegration::*;
        //匹配启动时候配置的存储方式，转到真正的实现去，这里只看文件的
        match (&self.0, location) {
             ...省略

           //文件存储 
            (File(file), path::Path::File(location)) => file
                .put(location, bytes, length)
                .await
                .context(FileObjectStoreError)?,
            _ => unreachable!(),
        }

        Ok(())
    }

//为File实现了ObjectStoreApi trait，相当于文件存储时候的实际实现
async fn put<S>(&self, location: &Self::Path, bytes: S, length: Option<usize>) -> Result<()>
    where
        S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
    {
        //读取之前ReadFilterResultsStream里的所有数据到content里
        let content = bytes
            .map_ok(|b| bytes::BytesMut::from(&b[..]))
            .try_concat()
            .await
            .context(UnableToStreamDataIntoMemory)?;
        //这里就是一个验证长度否则报错DataDoesNotMatchLength。宏编程，不用关注
        if let Some(length) = length {
            ensure!(
                content.len() == length,
                DataDoesNotMatchLength {
                    actual: content.len(),
                    expected: length,
                }
            );
        }
        //获取文件路径，就是启动时候配置的根路径加上数据路径
        let path = self.path(location);
        //创建这个文件出来
        let mut file = match fs::File::create(&path).await {
            Ok(f) => f,
            //如果是没有找到父路径，那就从新创建一次
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
                let parent = path
                    .parent()
                    .context(UnableToCreateFile { path: &path, err })?;
                fs::create_dir_all(&parent)
                    .await
                    .context(UnableToCreateDir { path: parent })?;

                match fs::File::create(&path).await {
                    Ok(f) => f,
                    Err(err) => return UnableToCreateFile { path, err }.fail(),
                }
            }
            //否则就失败了
            Err(err) => return UnableToCreateFile { path, err }.fail(),
        };
        //这里就是拷贝所有数据到这个文件中去
        tokio::io::copy(&mut &content[..], &mut file)
            .await
            .context(UnableToCopyDataToFile)?;
        //大功告成
        Ok(())
    }