Method open

src/persist/src/postgres.rs:275–351

Open a Postgres [Consensus] instance with `config`, for the collection named `shard`.

(config: PostgresConsensusConfig)

    /// Open a Postgres [Consensus] instance with `config`, for the collection
    /// named `shard`.
    pub async fn open(config: PostgresConsensusConfig) -> Result<Self, ExternalError> {
        // don't need to unredact here because we just want to pull out the username
        let pg_config: Config = config.url.to_string().parse()?;
        let role = pg_config.get_user().expect("failed to get PostgreSQL user");
        let create_schema = format!(
            "CREATE SCHEMA IF NOT EXISTS consensus AUTHORIZATION {}",
            escape_identifier(role),
        );

        let dyncfg = Arc::clone(&config.dyncfg);

        // Filled in below once we've detected the backend. The isolation resolver runs per
        // connection, so the flag takes effect as the pool cycles connections.
        let is_pg_backend = Arc::new(AtomicBool::new(false));
        let client_config = PostgresClientConfig::new(config.url, config.knobs, config.metrics)
            .with_isolation(Arc::new({
                let dyncfg = Arc::clone(&dyncfg);
                let is_pg_backend = Arc::clone(&is_pg_backend);
                move || {
                    let flag_enabled = USE_POSTGRES_TUNED_QUERIES.get(&dyncfg);
                    let is_pg_backend = is_pg_backend.load(Ordering::Relaxed);
                    if flag_enabled && is_pg_backend {
                        IsolationLevel::ReadCommitted
                    } else {
                        IsolationLevel::Serializable
                    }
                }
            }));
        let postgres_client = PostgresClient::open(client_config)?;

        let client = postgres_client.get_connection().await?;

        let mode = match pg_batch_execute(
            &client,
            &format!(
                "{}; {}{}; {};",
                create_schema, SCHEMA, CRDB_SCHEMA_OPTIONS, CRDB_CONFIGURE_ZONE,
            ),
        )
        .await
        {
            Ok(()) => PostgresMode::CockroachDB,
            Err(e) if e.code() == Some(&SqlState::INSUFFICIENT_PRIVILEGE) => {
                warn!(
                    "unable to ALTER TABLE consensus, this is expected and OK when connecting with a read-only user"
                );
                PostgresMode::CockroachDB
            }
            // Vanilla Postgres doesn't support the Cockroach zone configuration
            // that we attempted, so we use that to determine what mode we're in.
            Err(e)
                if e.code() == Some(&SqlState::INVALID_PARAMETER_VALUE)
                    || e.code() == Some(&SqlState::SYNTAX_ERROR) =>
            {
                info!(
                    "unable to initiate consensus with CRDB params, this is expected and OK when running against Postgres: {:?}",
                    e
                );