Method begin_statement_execution

src/adapter/src/statement_logging.rs:575–720

Begin statement execution logging from the frontend. (Corresponds to `Coordinator::begin_statement_execution`, which is used by the old peek sequencing.) This encapsulates all the statement logging setup: - Retrieves system config values - Performs sampling and throttling checks - Creates statement logging records - Attends to metrics. Returns None if the statement should not be logged (due to s

(
        &self,
        session: &mut Session,
        params: &Params,
        logging: &Arc<QCell<PreparedStatementLoggingInfo>>,
        system_config: &SystemVars,
        lifecycle_timestamps: O

    /// The `Option<PreparedStatementEvent>` is None when we have already logged the prepared
    /// statement before, and this is just a subsequent execution.
    pub fn begin_statement_execution(
        &self,
        session: &mut Session,
        params: &Params,
        logging: &Arc<QCell<PreparedStatementLoggingInfo>>,
        system_config: &SystemVars,
        lifecycle_timestamps: Option<LifecycleTimestamps>,
    ) -> Option<(
        StatementLoggingId,
        StatementBeganExecutionRecord,
        Row,
        Option<PreparedStatementEvent>,
    )> {
        // Skip logging for internal users unless explicitly enabled
        let enable_internal_statement_logging = system_config.enable_internal_statement_logging();
        if session.user().is_internal() && !enable_internal_statement_logging {
            return None;
        }

        let sample_rate = effective_sample_rate(session, system_config);

        let use_reproducible_rng = system_config.statement_logging_use_reproducible_rng();
        let target_data_rate: Option<u64> = system_config
            .statement_logging_target_data_rate()
            .map(|rate| rate.cast_into());
        let max_data_credit: Option<u64> = system_config
            .statement_logging_max_data_credit()
            .map(|credit| credit.cast_into());

        // Only lock the RNG when we actually need reproducible sampling (tests only)
        let sample = if use_reproducible_rng {
            let mut rng = self.reproducible_rng.lock().expect("rng lock poisoned");
            should_sample_statement(sample_rate, Some(&mut *rng))
        } else {
            should_sample_statement(sample_rate, None)
        };

        let sampled_label = sample.then_some("true").unwrap_or("false");
        session
            .metrics()
            .statement_logging_records(&[sampled_label])
            .inc_by(1);

        // Clone only the metrics needed below, before the mutable borrow of session.
        let unsampled_bytes_metric = session
            .metrics()
            .statement_logging_unsampled_bytes()
            .clone();
        let actual_bytes_metric = session.metrics().statement_logging_actual_bytes().clone();

        // Handle the accounted flag and record byte metrics
        let is_new_prepared_statement = if let Some((sql, accounted)) =
            match session.qcell_rw(logging) {
                PreparedStatementLoggingInfo::AlreadyLogged { .. } => None,
                PreparedStatementLoggingInfo::StillToLog { sql, accounted, .. } => {
                    Some((sql, accounted))
                }
            } {