时序数据库Influx-IOx源码学习九（查询主流程）

 let mut query = flight::Client::new(connection)
        .perform_query("databaseName", "select * from myMeasurement")
        .await
        .expect("query request should work");

async fn do_get(
        &self,
        //这个Ticket里就是保存的perform_query方法中封装的json数据
        request: Request<Ticket>,
    ) -> Result<Response<Self::DoGetStream>, tonic::Status> {
        //这里就是把json还原回来
        let ticket = request.into_inner();
        let json_str = String::from_utf8(ticket.ticket.to_vec()).context(InvalidTicket {
            ticket: ticket.ticket,
        })?;
        //反序列化成了ReadInfo结构
        let read_info: ReadInfo =
            serde_json::from_str(&json_str).context(InvalidQuery { query: &json_str })?;
        //拿到客户端设置的数据库名字
        let database = DatabaseName::new(&read_info.database_name).context(InvalidDatabaseName)?;
        //从内存中查找是否存在这个database名字，如果不存在就会报DatabaseNotFound错误回去
        //这里就是创建数据库的时候写入到内存里的
        //同时还应该记得iox的数据库必须一个节点创建一次。。hhhhha
        let db = self.server.db(&database).context(DatabaseNotFound {
            database_name: &read_info.database_name,
        })?;
        //这个是拿到之前创建数据库时候设置的线程池，可以回去参考第五章
        let executor = db.executor();
        //这里是创建出sql语句对应的physical_plan，后面再看
        let physical_plan = Planner::new(Arc::clone(&executor))
            .sql(db, &read_info.sql_query)
            .await
            .context(Planning)?;

        //使用线程异步的执行查询
        let results = executor
             //复制一下执行时候需要用到的信息
            .new_context()
             //真正的去执行
            .collect(Arc::clone(&physical_plan))
            .await
            .map_err(|e| Box::new(e) as _)
            .context(Query {
                database_name: &read_info.database_name,
            })?;

        //在写入的章节里应该知道了在RBChunk里面存储的是Arrow格式的。
        //在这个方法中就是调用arrow_flight工具包的方法，先把schema序列化到flight_buffer中
        let options = arrow::ipc::writer::IpcWriteOptions::default();
        let schema = physical_plan.schema();
        let schema_flight_data =
            arrow_flight::utils::flight_data_from_arrow_schema(schema.as_ref(), &options);

        let mut flights: Vec<Result<FlightData, tonic::Status>> = vec![Ok(schema_flight_data)];
        //上面得到的结果集，这里进行遍历，封装为要返回的数据结构
        let mut batches: Vec<Result<FlightData, tonic::Status>> = results
            .iter()
            //这个是为了给下面flight_data_from_arrow_batch这个方法打补丁用的
            //因为这个方法即便对于切片类型的batch也是盲目的序列化所有数据
            .map(optimize_record_batch)
            .collect::<Result<Vec<_>, Error>>()?
            .iter()
            //这里就是一条一条的把数据序列化到缓冲区里
            .flat_map(|batch| {
                let (flight_dictionaries, flight_batch) =
                    arrow_flight::utils::flight_data_from_arrow_batch(&batch, &options);
                //把数据包装在Result中
                flight_dictionaries
                    .into_iter()
                    .chain(std::iter::once(flight_batch))
                    .map(Ok)
            })
            .collect();

        //前面是schema，后面是数据
        flights.append(&mut batches);
        //返回一个数据的异步stream，有可能调用一次next就会释放一次cpu？
        let output = futures::stream::iter(flights);
        //数据以flight形式发送到了客户端，客户端先读取schema再读取数据。
        Ok(Response::new(Box::pin(output) as Self::DoGetStream))
    }