时序数据库Influx-IOx源码学习五（创建数据库）

enum Command {
    Convert { // 省略 ...},
    Meta {// 省略 ...},
    Database(commands::database::Config),
    Run(Box<commands::run::Config>),
    Stats(commands::stats::Config),
    Server(commands::server::Config),
    Writer(commands::writer::Config),
    Operation(commands::operations::Config),
}

pub struct Config {
    #[structopt(subcommand)]
    command: Command,
}
//见名知意，基本猜测一下就行了，慢慢使用到再回来看
enum Command {
    Create(Create),
    List(List),
    Get(Get),
    Write(Write),
    Query(Query),
    Chunk(chunk::Config),
    Partition(partition::Config),
}

Command::Create(command) => {
            //创建一个grpc的client
            let mut client = management::Client::new(connection);
            //设置基本的配置项
            let rules = DatabaseRules {
                //数据库名字
                name: command.name,
                //内存的各种配置，包含缓存大小，时间等等
                lifecycle_rules: Some(LifecycleRules {
                    //省略。。
                }),
                //设置分区的策略
                partition_template: Some(PartitionTemplate {
                    //省略。。
                }),

                 //其它都填充default
                ..Default::default()
            };
            //使用配置信息创建数据库，这里是生成了一个CreateDatabaseRequest去调用了远程服务器的方法
            client.create_database(rules).await?;

            println!("Ok");
        }

#[tonic::async_trait]
impl<M> management_service_server::ManagementService for ManagementService<M>
where
    M: ConnectionManager + Send + Sync + Debug + 'static,
{
    //省略其它方法。。。

 async fn create_database(
        &self,
        //这里就是接收CreateDatabaseRequest的请求
        request: Request<CreateDatabaseRequest>,
    ) -> Result<Response<CreateDatabaseResponse>, Status> {

         //对数据进行一下校验，然后获得在上面配置的rules规则
        let rules: DatabaseRules = request
            .into_inner()
            .rules
            .ok_or_else(|| FieldViolation::required(""))
            .and_then(TryInto::try_into)
            .map_err(|e| e.scope("rules"))?;

        //这里就是在第三章中提到的server_id，如果没配置就会报错了
        let server_id = match self.server.require_id().ok() {
            Some(id) => id,
            None => return Err(NotFound::default().into()),
        };
        //这里就是真正的去创建，在下面继续跟踪
        match self.server.create_database(rules, server_id).await {
            Ok(_) => Ok(Response::new(CreateDatabaseResponse {})),
            Err(Error::DatabaseAlreadyExists { db_name }) => {
                return Err(AlreadyExists {
                    resource_type: "database".to_string(),
                    resource_name: db_name,
                    ..Default::default()
                }
                .into())
            }
            Err(e) => Err(default_server_error_handler(e)),
        }
    }
}

pub async fn create_database(&self, rules: DatabaseRules, server_id: NonZeroU32) -> Result<()> {
        //检查server_id
        self.require_id()?;
        //把数据库名字存储到内存中，最终保存到一个btreemap中
        let db_reservation = self.config.create_db(rules)?;
        //对数据进行持久化保存
        self.persist_database_rules(db_reservation.rules().clone())
            .await?;
        //启动数据库后台线程，在内存中写入数据库状态
        db_reservation.commit(server_id, Arc::clone(&self.store), Arc::clone(&self.exec));

        Ok(())
    }

pub async fn persist_database_rules<'a>(&self, rules: DatabaseRules) -> Result<()> {
        //生成一个新的数据库路径
        let location = object_store_path_for_database_config(&self.root_path()?, &rules.name);
        //序列化DatabaseRules这个pb到byte流
        let mut data = BytesMut::new();
        rules.encode(&mut data).context(ErrorSerializing)?;
        let len = data.len();
        let stream_data = std::io::Result::Ok(data.freeze());
        //将pb的内容进行存储
        self.store
            .put(
                &location,
                futures::stream::once(async move { stream_data }),
                Some(len),
            )
            .await
            .context(StoreError)?;
        Ok(())
    }

./influxdb_iox run --writer-id 1 --object-store file --data-dir ~/influxtest/
./influxdb_iox database create test

$ ./scripts/prototxt decode influxdata.iox.management.v1.DatabaseRules \
    < ~/influxtest/1/test/rules.pb

influxdata/iox/management/v1/service.proto:6:1: warning: Import google/protobuf/field_mask.proto is unused.
name: "test"
partition_template {
  parts {
    time: "%Y-%m-%d %H:00:00"
  }
}
lifecycle_rules {
  mutable_linger_seconds: 300
  mutable_size_threshold: 10485760
  buffer_size_soft: 52428800
  buffer_size_hard: 104857600
  sort_order {
    order: ORDER_ASC
    created_at_time {
    }
  }
}