Method create_table_impl

crates/storage-postgres/src/create_table.rs:17–392

Core implementation of `create_table` (Fix #4: wrapped in a transaction).

(
        &self,
        account_id: &str,
        input: CreateTableInput,
    )

impl PostgresEngine {
    /// Core implementation of `create_table` (Fix #4: wrapped in a transaction).
    pub(crate) async fn create_table_impl(
        &self,
        account_id: &str,
        input: CreateTableInput,
    ) -> Result<TableDescription, StorageError> {
        Self::validate_account_id(account_id)?;
        let table_id = uuid::Uuid::new_v4().to_string();
        let table_arn = table_arn(&self.region, account_id, &input.table_name);
        let billing_mode = input.billing_mode.unwrap_or(BillingMode::Provisioned);
        let key_schema_json = serde_json::to_value(&input.key_schema)
            .map_err(|e| StorageError::Internal(e.to_string()))?;
        let attr_defs_json = serde_json::to_value(&input.attribute_definitions)
            .map_err(|e| StorageError::Internal(e.to_string()))?;
        let billing_str = match billing_mode {
            BillingMode::Provisioned => "PROVISIONED",
            BillingMode::PayPerRequest => "PAY_PER_REQUEST",
        };
        // Fix #7: Use serde_json::to_value directly instead of redundant closures
        let pt_json = input
            .provisioned_throughput
            .as_ref()
            .map(serde_json::to_value)
            .transpose()
            .map_err(|e| StorageError::Internal(e.to_string()))?;
        let stream_json = input
            .stream_specification
            .as_ref()
            .map(serde_json::to_value)
            .transpose()
            .map_err(|e| StorageError::Internal(e.to_string()))?;
        let deletion_protection = input.deletion_protection_enabled.unwrap_or(false);

        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| StorageError::Internal(e.to_string()))?;

        // Insert table metadata, returning creation timestamp and actual status.
        // Use PG error code 23505 for robust duplicate detection instead of string matching.
        // H-5: Insert as CREATING with a scheduled transition to ACTIVE,
        // or directly as ACTIVE when control_plane_delay_seconds=0 (no async
        // transition needed). This lets external test suites that don't call
        // waitForActive() work correctly.
        let (creation_epoch, actual_status): (f64, String) = sqlx::query_as(
            r"WITH delay AS (
                SELECT COALESCE(
                  (SELECT value::FLOAT8 FROM settings WHERE key = 'control_plane_delay_seconds'), 0.25
                ) AS secs
              )
              INSERT INTO tables
               (account_id, table_name, key_schema, attribute_definitions, billing_mode,
                provisioned_throughput, stream_specification, table_status,
                creation_date_time, table_arn, table_id, deletion_protection_enabled,
                status_transition_at)
               VALUES ($1, $2, $3, $4, $5, $6, $7,
                CASE WHEN (SELECT secs FROM delay) = 0
                     THEN 'ACTIVE' ELSE 'CREATING' END,