时序数据库Influx-IOx源码学习七（Chunk的生命周期）

//这里需要注意，这些变体里的Chunk结构都是不相同的
//也就是有内存数据拷贝的工作
pub enum ChunkState {
    //内部移动数据时候用的
    Invalid,
    //可以写入
    Open(MBChunk),
    //还能继续写入，但很快会被关闭
    Closing(MBChunk),
    //已经不能写入了，准备移动到readbuffer
    Moving(Arc<MBChunk>),
    //已经被移动到了read buffer
    Moved(Arc<ReadBufferChunk>),
    //准备写入持久化存储
    WritingToObjectStore(Arc<ReadBufferChunk>),
    //写入持久化存储完成
    WrittenToObjectStore(Arc<ReadBufferChunk>, Arc<ParquetChunk>),
}

    //后台线程的方法入口，在创建完成数据库后，就会调用到这个方法
    pub async fn background_worker(
        self: &Arc<Self>,
        shutdown: tokio_util::sync::CancellationToken,
    ) {
        //创建一个定时器，周期性的执行
        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(1));
        let mut lifecycle_manager = LifecycleManager::new(Arc::clone(&self));
        //没有收到停止服务器时候的信号就一直执行，1秒一次
        while !shutdown.is_cancelled() {
            //记录执行的次数，每次加1，Ordering::Relaxed代表的单线程里的原子操作
            self.worker_iterations.fetch_add(1, Ordering::Relaxed);
            //进入生命周期的管理
            lifecycle_manager.check_for_work();
            //收到不同信号之后的处理方法
            tokio::select! {
                _ = interval.tick() => {},
                _ = shutdown.cancelled() => break
            }
        }

        info!("finished background worker");
    }

fn check_for_work(&mut self, now: DateTime<Utc>) {
        //获取创建数据库的时候，对于Chunk的相关配置
        let rules = self.rules();
        //根据配置的排序规则，获取出内存里所有的chunk
        let chunks = self.chunks(&rules.sort_order);

        let mut buffer_size = 0;

        //判断是不是有其他的任务正在执行，move我理解针对于read buffer，write对于持久化
        let mut move_active = self.is_move_active();
        let mut write_active = self.is_write_active();

         //遍历所有块，检查哪些块可以被持久化
        for chunk in &chunks {
            //获取当前chunk的锁
            let chunk_guard = chunk.upgradable_read();
            //获取chunk占用的内存大小
            buffer_size += Self::chunk_size(&*chunk_guard);
            //没有移动任务并且Chunk里最后的写入时间比较老
            let would_move = !move_active && can_move(&rules, &*chunk_guard, now);
            //没有写出任务，并且开启了持久化
            let would_write = !write_active && rules.persist;

            //判断chunk的生命周期
            match chunk_guard.state() {
                 //属于open状态，并且是需要移动的（上面的逻辑里有展示什么是需要移动的）
                 //这里我理解就是相当于实时写入时候的一个补充方案
                 //试想，如果一个chunk一直不写入数据，可能有一年了，查询都不再用这些数据了，内存却被一直占用
                ChunkState::Open(_) if would_move => {
                    let mut chunk_guard = RwLockUpgradableReadGuard::upgrade(chunk_guard);
                    //切换状态到closing
                    chunk_guard.set_closing().expect("cannot close open chunk");
                    let partition_key = chunk_guard.key().to_string();
                    let chunk_id = chunk_guard.id();
                    std::mem::drop(chunk_guard);
                    move_active = true;
                    //移动到read_buffer，变为不可写入状态（启动了一个异步的线程，后面看）
                    self.move_to_read_buffer(partition_key, chunk_id);
                }
                //这里有几种情况，同样会在别处触发为closing
                //例如：chunk大小超过了设置的可变内存大小的时候
                ChunkState::Closing(_) if would_move => {
                    let partition_key = chunk_guard.key().to_string();
                    let chunk_id = chunk_guard.id();
                    std::mem::drop(chunk_guard);
                    move_active = true;
                    //移动到read_buffer
                    self.move_to_read_buffer(partition_key, chunk_id);
                }
                //已经被挪动到readbuffer中的
                ChunkState::Moved(_) if would_write => {
                    let partition_key = chunk_guard.key().to_string();
                    let chunk_id = chunk_guard.id();
                    std::mem::drop(chunk_guard);
                    write_active = true;
                    //写入到对象存储
                    self.write_to_object_store(partition_key, chunk_id);
                }
                _ => {}
            }
        }
        //这里是主要检查内存限制的逻辑，当所有chunk的大小超过限制的时候就要清理Chunk
        if let Some(soft_limit) = rules.buffer_size_soft {
            let mut chunks = chunks.iter();

            while buffer_size > soft_limit.get() {
                match chunks.next() {
                    Some(chunk) => {
                        //获取读锁
                        let chunk_guard = chunk.read();
                        //如果配置了可以清理未持久化数据，那么处在read_buffer里的数据也会被清理
                        //一定会清理已经被持久化到对象存储上的数据
                        if (rules.drop_non_persisted
                            && matches!(chunk_guard.state(), ChunkState::Moved(_)))
                            || matches!(chunk_guard.state(), ChunkState::WrittenToObjectStore(_, _))
                        {
                            let partition_key = chunk_guard.key().to_string();
                            let chunk_id = chunk_guard.id();
                            buffer_size =
                                buffer_size.saturating_sub(Self::chunk_size(&*chunk_guard));
                            std::mem::drop(chunk_guard);
                            //真真正正的删除逻辑后面看
                            self.drop_chunk(partition_key, chunk_id)
                        }
                    }
                    //没有什么可以释放的了
                    None => {
                        warn!(db_name=self.db_name(), soft_limit, buffer_size,
                              "soft limited exceeded, but no chunks found that can be evicted. Check lifecycle rules");
                        break;
                    }
                }
            }
        }
    }

pub async fn load_chunk_to_read_buffer(
        &self,
        partition_key: &str,
        chunk_id: u32,
    ) -> Result<Arc<DbChunk>> {
        //根据partition_key及chunk_id获取内存中存储的Chunk
        let chunk = {
            let partition = self
                .catalog
                .valid_partition(partition_key)
                .context(LoadingChunk {
                    partition_key,
                    chunk_id,
                })?;
            let partition = partition.read();

            partition.chunk(chunk_id).context(LoadingChunk {
                partition_key,
                chunk_id,
            })?
        };

        //设置当前的Chunk为Moving状态
        let mb_chunk = {
            let mut chunk = chunk.write();
            chunk.set_moving().context(LoadingChunk {
                partition_key,
                chunk_id,
            })?
        };
        info!(%partition_key, %chunk_id, "chunk marked MOVING, loading tables into read buffer");
        let mut batches = Vec::new();
        //这里是拿到Chunk中每个Cloumn的统计信息，分别是min，max，count
        let table_stats = mb_chunk.table_summaries();

        //从新创建一个ReadBufferChunk，后面准备把所有数据都拷贝到这里
        //还需要告诉内存管理这里新申请了多少空间
        let rb_chunk =
            ReadBufferChunk::new_with_memory_tracker(chunk_id, &self.memory_registries.read_buffer);

        for stats in table_stats {
            //把内存中的数据，全部重新拷贝一次，转换为arrow格式
            mb_chunk
                .table_to_arrow(&mut batches, &stats.name, Selection::All)
                //这里应该是还没有写完，如果出现错误，这个Chunk该怎么处理？
                .expect("Loading chunk to mutable buffer");
            //循环拷贝
            for batch in batches.drain(..) {
                rb_chunk.upsert_table(&stats.name, batch)
            }
        }

        let mut chunk = chunk.write();
        //更新写入缓存里的Chunk为Moved状态，同时Chunk内容修改为了ReadBuffer的Chunk
        //对于Chunk的结构后面看
        chunk.set_moved(Arc::new(rb_chunk)).context(LoadingChunk {
            partition_key,
            chunk_id,
        })?;

        //工作全部都完成了,调用做快照的方法,方法里什么都没做,返回新Chunk的一个Arc指针
        Ok(DbChunk::snapshot(&chunk))
    }

//主要是存储一个数据块的描述信息，名字、最后写入时间等
Server::db::catalog::chunk
//数据从客户端直接写入的内存块
mutable_buffer::chunk
//在moving时候拷贝的新数据块，arrow结构
read_buffer::chunk
//parquet对应的chunk
parquet_file::chunk
//query模块下对PartitionChunk重新命名了一下
//对于相同的partition key的数据抽象的行为
query -> type Chunk: PartitionChunk;
//实现PartitionChunk定义的方法，对不同位置下的chunk的操作
//如ParquetFile、MutableBuffer等
server::db::chunk