sqlx-otel 0.3.0

#![allow(dead_code, clippy::must_use_candidate, clippy::missing_panics_doc)]

use opentelemetry::trace::{SpanKind, Status};
use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider};
use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider, SpanData};
use sqlx_otel::QueryAnnotations;

/// Test harness that installs in-memory span and metric exporters as the global providers,
/// collects telemetry in-process, and cleans up on drop.
pub struct TestTelemetry {
    span_exporter: InMemorySpanExporter,
    metric_exporter: InMemoryMetricExporter,
    tracer_provider: SdkTracerProvider,
    meter_provider: SdkMeterProvider,
}

impl TestTelemetry {
    /// Install in-memory exporters as the global tracer and meter providers.
    #[must_use]
    pub fn install() -> Self {
        let span_exporter = InMemorySpanExporter::default();
        let tracer_provider = SdkTracerProvider::builder()
            .with_simple_exporter(span_exporter.clone())
            .build();
        opentelemetry::global::set_tracer_provider(tracer_provider.clone());

        let metric_exporter = InMemoryMetricExporter::default();
        let reader = PeriodicReader::builder(metric_exporter.clone()).build();
        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
        opentelemetry::global::set_meter_provider(meter_provider.clone());

        Self {
            span_exporter,
            metric_exporter,
            tracer_provider,
            meter_provider,
        }
    }

    /// Return all finished spans, flushing the provider first.
    #[must_use]
    pub fn spans(&self) -> Vec<SpanData> {
        let _ = self.tracer_provider.force_flush();
        self.span_exporter.get_finished_spans().unwrap_or_default()
    }

    /// Return all finished metrics, flushing the provider first.
    #[must_use]
    pub fn metrics(&self) -> Vec<opentelemetry_sdk::metrics::data::ResourceMetrics> {
        let _ = self.meter_provider.force_flush();
        self.metric_exporter
            .get_finished_metrics()
            .unwrap_or_default()
    }

    /// Drain the in-memory exporters so the next call to [`spans`](Self::spans) or
    /// [`metrics`](Self::metrics) sees a fresh window.
    ///
    /// Use this between sections of a single test that want to assert on a bounded set of
    /// spans/metrics, instead of re-installing the global telemetry providers (which is
    /// racy: a fresh `install()` mid-test replaces the global tracer/meter providers and
    /// silently changes which exporter receives subsequent operations).
    pub fn reset(&self) {
        let _ = self.tracer_provider.force_flush();
        let _ = self.meter_provider.force_flush();
        self.span_exporter.reset();
        self.metric_exporter.reset();
    }
}

impl Drop for TestTelemetry {
    fn drop(&mut self) {
        let _ = self.tracer_provider.shutdown();
        let _ = self.meter_provider.shutdown();
    }
}

// ---------------------------------------------------------------------------
// Shared assertion helpers
// ---------------------------------------------------------------------------

/// Find the attribute value for a given key in a span.
pub fn attr(span: &SpanData, key: &str) -> Option<opentelemetry::Value> {
    span.attributes
        .iter()
        .find(|kv| kv.key.as_str() == key)
        .map(|kv| kv.value.clone())
}

/// Find the attribute value for a given key on a histogram data point.
///
/// Mirrors [`attr`] for the metric side: walks the data point's attribute iterator and
/// returns the cloned value for the first matching key. Used by the metric-attribute
/// assertions in `test_operation_duration_metric_carries_*` to confirm the histogram
/// emits the same dimensions the span carries.
pub fn metric_attr(
    dp: &opentelemetry_sdk::metrics::data::HistogramDataPoint<f64>,
    key: &str,
) -> Option<opentelemetry::Value> {
    dp.attributes()
        .find(|kv| kv.key.as_str() == key)
        .map(|kv| kv.value.clone())
}

/// Find a single histogram data point on the named metric whose attribute set contains
/// the given expected key/value pairs. Returns `None` if the metric is absent or no data
/// point matches. Generalises [`find_duration_data_point_with`] to any histogram (e.g.
/// `db.client.response.affected_rows`).
pub fn find_histogram_data_point_with(
    metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics],
    metric_name: &str,
    expected: &[(&str, &str)],
) -> Option<opentelemetry_sdk::metrics::data::HistogramDataPoint<f64>> {
    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
    for rm in metrics {
        for sm in rm.scope_metrics() {
            for metric in sm.metrics() {
                if metric.name() != metric_name {
                    continue;
                }
                if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() {
                    for dp in hist.data_points() {
                        let matches = expected.iter().all(|(k, v)| {
                            dp.attributes().any(|kv| {
                                kv.key.as_str() == *k
                                    && matches!(
                                        &kv.value,
                                        opentelemetry::Value::String(s) if s.as_str() == *v
                                    )
                            })
                        });
                        if matches {
                            return Some(dp.clone());
                        }
                    }
                }
            }
        }
    }
    None
}

/// Assert the `db.client.response.affected_rows` histogram has at least one data point
/// for the given backend `system`. The exact recorded value is already pinned on the
/// span via `db.response.affected_rows`; this helper checks that the metric mirror is
/// reaching the meter (the in-memory exporter aggregates cumulatively, so per-call value
/// assertions across `tel.reset()` boundaries are not robust).
pub fn assert_affected_rows_metric(tel: &TestTelemetry, system: &str) {
    let metrics = tel.metrics();
    let dp = find_histogram_data_point_with(
        &metrics,
        "db.client.response.affected_rows",
        &[("db.system.name", system)],
    )
    .unwrap_or_else(|| {
        panic!("no db.client.response.affected_rows data point found for system {system:?}")
    });
    assert!(
        dp.count() > 0,
        "db.client.response.affected_rows data point has zero count",
    );
}

/// Locate the `db.client.operation.duration` histogram in a `ResourceMetrics` snapshot and
/// return the first data point.
///
/// Returns `None` if the metric is absent or has no data points. Tests that assert on the
/// data point's attributes should `unwrap()` the result so absence fails the test loudly
/// rather than silently passing an empty-attribute set.
pub fn find_duration_data_point(
    metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics],
) -> Option<opentelemetry_sdk::metrics::data::HistogramDataPoint<f64>> {
    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
    for rm in metrics {
        for sm in rm.scope_metrics() {
            for metric in sm.metrics() {
                if metric.name() != "db.client.operation.duration" {
                    continue;
                }
                if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() {
                    if let Some(dp) = hist.data_points().next() {
                        return Some(dp.clone());
                    }
                }
            }
        }
    }
    None
}

/// Find the first `db.client.operation.duration` data point whose attribute set contains
/// every `(key, value)` pair in `expected`. Each expected value is matched as a string
/// `opentelemetry::Value::String`. Returns `None` when no matching data point exists.
///
/// Used by [`assert_metric_data_point`] and the per-method macros to verify that
/// instrumentation for a specific scenario landed on the histogram with the dimensions
/// the test asserts the *span* carries – i.e. metric/span attribute parity.
pub fn find_duration_data_point_with(
    metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics],
    expected: &[(&str, &str)],
) -> Option<opentelemetry_sdk::metrics::data::HistogramDataPoint<f64>> {
    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
    for rm in metrics {
        for sm in rm.scope_metrics() {
            for metric in sm.metrics() {
                if metric.name() != "db.client.operation.duration" {
                    continue;
                }
                if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() {
                    for dp in hist.data_points() {
                        let matches = expected.iter().all(|(k, v)| {
                            dp.attributes().any(|kv| {
                                kv.key.as_str() == *k
                                    && matches!(
                                        &kv.value,
                                        opentelemetry::Value::String(s) if s.as_str() == *v
                                    )
                            })
                        });
                        if matches {
                            return Some(dp.clone());
                        }
                    }
                }
            }
        }
    }
    None
}

/// Assert that the `db.client.operation.duration` histogram has a data point matching
/// every `(key, value)` pair in `expected`, and that its `count() > 0`. Panics with a
/// helpful message naming the missing pairs when no data point matches – the panic body
/// is intentionally verbose so assertion failures point at the unsatisfied dimension
/// rather than a generic "metric missing".
pub fn assert_metric_data_point(tel: &TestTelemetry, expected: &[(&str, &str)]) {
    let metrics = tel.metrics();
    let dp = find_duration_data_point_with(&metrics, expected).unwrap_or_else(|| {
        panic!(
            "no db.client.operation.duration data point found matching expected attrs {expected:?}",
        )
    });
    assert!(
        dp.count() > 0,
        "matching data point has zero count for expected attrs {expected:?}",
    );
}

/// Assert that the duration histogram recorded a data point for the given backend
/// `system`. The minimum bar every per-method macro should clear: this is the metric
/// equivalent of [`assert_common_span_attributes`] and verifies that the method's
/// instrumentation reached the meter at all.
pub fn assert_metric_for_system(tel: &TestTelemetry, system: &str) {
    assert_metric_data_point(tel, &[("db.system.name", system)]);
}

/// Assert that the duration histogram recorded a data point matching the standard
/// [`test_annotations`] keys (`db.operation.name = "SELECT"`,
/// `db.collection.name = "users"`) plus `db.system.name`. Used by every macro that
/// exercises the annotated path so the metric carries the same dimensions the span
/// assertions check via [`assert_annotated_span`].
pub fn assert_annotated_metric(tel: &TestTelemetry, dialect: &Dialect) {
    assert_metric_data_point(
        tel,
        &[
            ("db.system.name", dialect.system),
            ("db.operation.name", "SELECT"),
            ("db.collection.name", "users"),
        ],
    );
}

/// Assert that the duration histogram recorded an *annotated* error-path data point
/// matching the standard [`test_annotations`] keys (`db.operation.name = "SELECT"`,
/// `db.collection.name = "users"`) plus a non-empty `error.type` plus the expected
/// `db.system.name`. Used by every `*_records_error` macro that exercises an annotated
/// failing call: pins the contract that `record_error` keeps annotation attrs and error
/// attrs on the same `Vec<KeyValue>` so a single histogram data point carries both
/// dimensions, which dashboards rely on to slice latency by failed operation + failed
/// collection.
pub fn assert_annotated_error_metric(tel: &TestTelemetry, dialect: &Dialect) {
    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
    let metrics = tel.metrics();
    let dp = metrics
        .iter()
        .flat_map(opentelemetry_sdk::metrics::data::ResourceMetrics::scope_metrics)
        .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
        .filter(|m| m.name() == "db.client.operation.duration")
        .filter_map(|m| {
            if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = m.data() {
                Some(hist)
            } else {
                None
            }
        })
        .flat_map(opentelemetry_sdk::metrics::data::Histogram::data_points)
        .find(|dp| {
            let attr_eq = |key: &str, value: &str| {
                dp.attributes().any(|kv| {
                    kv.key.as_str() == key
                        && matches!(
                            &kv.value,
                            opentelemetry::Value::String(s) if s.as_str() == value
                        )
                })
            };
            let attr_present = |key: &str| {
                dp.attributes().any(|kv| {
                    kv.key.as_str() == key
                        && matches!(
                            &kv.value,
                            opentelemetry::Value::String(s) if !s.as_str().is_empty()
                        )
                })
            };
            attr_eq("db.system.name", dialect.system)
                && attr_eq("db.operation.name", "SELECT")
                && attr_eq("db.collection.name", "users")
                && attr_present("error.type")
        });
    assert!(
        dp.is_some(),
        "no annotated db.client.operation.duration data point with error.type and \
         db.system.name = {:?} found",
        dialect.system,
    );
}

/// Assert that the duration histogram recorded a data point on the error path: the data
/// point must carry `error.type` (any non-empty string – the exact value depends on the
/// `sqlx::Error` variant under test) and the expected `db.system.name`. Used by every
/// `*_records_error` macro.
pub fn assert_error_metric(tel: &TestTelemetry, dialect: &Dialect) {
    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
    let metrics = tel.metrics();
    let dp = metrics
        .iter()
        .flat_map(opentelemetry_sdk::metrics::data::ResourceMetrics::scope_metrics)
        .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
        .filter(|m| m.name() == "db.client.operation.duration")
        .filter_map(|m| {
            if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = m.data() {
                Some(hist)
            } else {
                None
            }
        })
        .flat_map(opentelemetry_sdk::metrics::data::Histogram::data_points)
        .find(|dp| {
            let has_error_type = dp.attributes().any(|kv| {
                kv.key.as_str() == "error.type"
                    && matches!(&kv.value, opentelemetry::Value::String(s) if !s.as_str().is_empty())
            });
            let has_system = dp.attributes().any(|kv| {
                kv.key.as_str() == "db.system.name"
                    && matches!(
                        &kv.value,
                        opentelemetry::Value::String(s) if s.as_str() == dialect.system
                    )
            });
            has_error_type && has_system
        });
    assert!(
        dp.is_some(),
        "no db.client.operation.duration data point with error.type and db.system.name = {:?} found",
        dialect.system,
    );
}

/// Assert that a span carries the common attributes every instrumented operation must have.
///
/// `system` is the expected `db.system.name` value (e.g. `"sqlite"`, `"postgresql"`).
///
/// `db.namespace` and `db.query.text` are checked for non-empty string content rather
/// than mere presence, so a regression that emits empty strings or non-string values is
/// caught at the helper level.
pub fn assert_common_span_attributes(span: &SpanData, system: &str) {
    assert_eq!(span.span_kind, SpanKind::Client);
    assert_eq!(
        span.name, system,
        "span name should fall back to db.system.name"
    );
    assert_eq!(
        attr(span, "db.system.name"),
        Some(opentelemetry::Value::String(system.to_owned().into())),
        "db.system.name missing or wrong"
    );
    let namespace = attr(span, "db.namespace");
    assert!(
        matches!(&namespace, Some(opentelemetry::Value::String(s)) if !s.as_str().is_empty()),
        "db.namespace should be a non-empty string, got {namespace:?}",
    );
    let query_text = attr(span, "db.query.text");
    assert!(
        matches!(&query_text, Some(opentelemetry::Value::String(s)) if !s.as_str().is_empty()),
        "db.query.text should be a non-empty string, got {query_text:?}",
    );
}

// ---------------------------------------------------------------------------
// Shared fixtures
// ---------------------------------------------------------------------------

/// Row shape for the shared `macro_users` table used by the macro-form integration tests.
///
/// Generic over the primary-key type so each backend can pick the variant that matches its
/// column type (sqlite `INTEGER` decodes to `i64`; postgres `INTEGER` and mysql `INT`
/// decode to `i32`).
#[derive(Debug, sqlx::FromRow)]
pub struct MacroUser<Id> {
    pub id: Id,
    pub name: String,
}

/// Assert that a span has error status, an `error.type` attribute, and an exception event
/// with `exception.type` and `exception.message`.
///
/// All three attribute values are required to be non-empty strings, so a regression that
/// emits empty exception metadata is caught here rather than slipping past the suite.
pub fn assert_error_span(span: &SpanData) {
    assert!(
        matches!(&span.status, Status::Error { .. }),
        "span status should be Error, got {:?}",
        span.status
    );
    let error_type = attr(span, "error.type");
    assert!(
        matches!(&error_type, Some(opentelemetry::Value::String(s)) if !s.as_str().is_empty()),
        "error.type should be a non-empty string, got {error_type:?}",
    );
    let event = span
        .events
        .iter()
        .find(|e| e.name == "exception")
        .expect("exception event missing");
    let exception_type = event
        .attributes
        .iter()
        .find(|kv| kv.key.as_str() == "exception.type")
        .map(|kv| kv.value.clone());
    assert!(
        matches!(&exception_type, Some(opentelemetry::Value::String(s)) if !s.as_str().is_empty()),
        "exception.type should be a non-empty string, got {exception_type:?}",
    );
    let exception_message = event
        .attributes
        .iter()
        .find(|kv| kv.key.as_str() == "exception.message")
        .map(|kv| kv.value.clone());
    assert!(
        matches!(&exception_message, Some(opentelemetry::Value::String(s)) if !s.as_str().is_empty()),
        "exception.message should be a non-empty string, got {exception_message:?}",
    );
}

// ---------------------------------------------------------------------------
// Backend parameterisation
// ---------------------------------------------------------------------------
//
// Why macros and not generic functions: the library's `impl_executor!` macro at
// `src/executor.rs` instantiates `Executor` impls for `&Pool<DB>`, `&mut
// PoolConnection<DB>`, `&mut Transaction<'_, DB>`, and the matching `Annotated` /
// `AnnotatedMut` wrappers, each gated by `for<'a> &'a mut DB::Connection: Executor<'a,
// Database = DB>`. Test bodies generic over `DB` (with that HRTB declared) trigger
// trait-resolution overflow on stable rustc – the compiler tries to satisfy the bound
// against multiple wrapper impls and recurses. Bumping `recursion_limit` does not
// help; the chain genuinely diverges.
//
// `macro_rules!` sidesteps the issue entirely: each invocation expands at the call
// site with concrete types, so the bound chain resolves directly against the upstream
// `sqlx::Sqlite` / `sqlx::Postgres` / `sqlx::MySql` impls. Trade-off: error messages
// point at the expansion site instead of the macro definition; mitigated by keeping
// each macro short and well-documented.

/// Per-backend SQL fragments and metadata used by parameterised test bodies.
///
/// Test bodies that vary only in SQL syntax (column types, upsert form, string concat)
/// read the relevant field from a `&Dialect` argument instead of being duplicated across
/// three backend files. Each backend file passes the matching constant
/// (`SQLITE_DIALECT`, `POSTGRES_DIALECT`, `MYSQL_DIALECT`) to the shared body.
pub struct Dialect {
    /// The OpenTelemetry `db.system.name` value (`"sqlite"`, `"postgresql"`, `"mysql"`).
    pub system: &'static str,
    /// Column definition for an integer primary key in CREATE TABLE: e.g. `"INTEGER
    /// PRIMARY KEY"` for sqlite, `"INT PRIMARY KEY"` for postgres / mysql.
    pub id_pk_column: &'static str,
    /// Column definition for a non-null text column: e.g. `"TEXT NOT NULL"` for sqlite
    /// and postgres, `"VARCHAR(255) NOT NULL"` for mysql.
    pub text_column: &'static str,
    /// Full SQL for an upsert that updates `affected_test`'s row id=1 to a new name.
    /// Each backend's syntax differs (`INSERT OR REPLACE` / `ON CONFLICT … DO UPDATE` /
    /// `ON DUPLICATE KEY UPDATE`).
    pub upsert_sql: &'static str,
    /// Expected `db.response.affected_rows` for the upsert above. Sqlite and postgres
    /// report `1`; mysql reports `2` (it counts match + update).
    pub upsert_affected_rows: i64,
    /// Full SQL for an UPDATE that mutates two rows by appending `_updated` to `name`
    /// using the dialect's string-concat operator (`||` for sqlite/postgres, `CONCAT(...)`
    /// for mysql).
    pub string_concat_update_sql: &'static str,
    /// Full SQL of the form `SELECT (?1 + ?2) AS sum`, accepting two `i32` binds and
    /// returning an `i64` named `sum`. Each backend uses its own placeholder syntax and
    /// (for postgres / mysql) explicit casts so the result fits in `i64` uniformly.
    pub bind_two_sum_sql: &'static str,
    /// Full SQL of the form `SELECT <placeholder>` for `prepare_with` calls that supply
    /// no concrete binds. Each backend uses its own placeholder syntax (`?` for sqlite
    /// and mysql, `$1` for postgres).
    pub prepare_with_select_sql: &'static str,
}

pub const SQLITE_DIALECT: Dialect = Dialect {
    system: "sqlite",
    id_pk_column: "INTEGER PRIMARY KEY",
    text_column: "TEXT NOT NULL",
    upsert_sql: "INSERT OR REPLACE INTO affected_test (id, name) VALUES (1, 'alice_updated')",
    upsert_affected_rows: 1,
    string_concat_update_sql: "UPDATE affected_test SET name = name || '_updated' WHERE id IN (2, 3)",
    bind_two_sum_sql: "SELECT ?1 + ?2 AS sum",
    prepare_with_select_sql: "SELECT ?",
};

pub const POSTGRES_DIALECT: Dialect = Dialect {
    system: "postgresql",
    id_pk_column: "INT PRIMARY KEY",
    text_column: "TEXT NOT NULL",
    upsert_sql: "INSERT INTO affected_test (id, name) VALUES (1, 'alice_updated') \
                 ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name",
    upsert_affected_rows: 1,
    string_concat_update_sql: "UPDATE affected_test SET name = name || '_updated' WHERE id IN (2, 3)",
    bind_two_sum_sql: "SELECT ($1::bigint + $2::bigint) AS sum",
    prepare_with_select_sql: "SELECT $1",
};

pub const MYSQL_DIALECT: Dialect = Dialect {
    system: "mysql",
    id_pk_column: "INT PRIMARY KEY",
    text_column: "VARCHAR(255) NOT NULL",
    upsert_sql: "INSERT INTO affected_test (id, name) VALUES (1, 'alice_updated') \
                 ON DUPLICATE KEY UPDATE name = VALUES(name)",
    // MySQL counts ON DUPLICATE KEY UPDATE as match (1) + update (1) = 2.
    upsert_affected_rows: 2,
    string_concat_update_sql: "UPDATE affected_test SET name = CONCAT(name, '_updated') WHERE id IN (2, 3)",
    bind_two_sum_sql: "SELECT CAST(? + ? AS SIGNED) AS sum",
    prepare_with_select_sql: "SELECT ?",
};

/// `DROP TABLE IF EXISTS` then `CREATE TABLE` at the supplied pool. Used at the top of
/// every parameterised body so sqlite (fresh `:memory:` per pool) and postgres / mysql
/// (shared container) behave identically. Expanded inline at the call site via the
/// `fresh_table!` macro so it operates on concrete pool types without HRTB issues.
#[macro_export]
macro_rules! fresh_table {
    ($pool:expr, $table:expr, $columns:expr) => {{
        use sqlx::Executor as _;
        let drop_sql = format!("DROP TABLE IF EXISTS {}", $table);
        $pool.execute(drop_sql.as_str()).await.unwrap();
        let create_sql = format!("CREATE TABLE {} ({})", $table, $columns);
        $pool.execute(create_sql.as_str()).await.unwrap();
    }};
}

/// The standard annotation set used across most annotation tests:
/// `db.operation.name = "SELECT"`, `db.collection.name = "users"`.
pub fn test_annotations() -> QueryAnnotations {
    QueryAnnotations::new()
        .operation("SELECT")
        .collection("users")
}

/// Assert that a span carries the attributes set by [`test_annotations`].
///
/// The `db.system.name` value is taken from the supplied `Dialect`, so this helper works
/// for every backend without per-file duplication.
pub fn assert_annotated_span(span: &SpanData, dialect: &Dialect) {
    assert_eq!(span.span_kind, SpanKind::Client);
    assert_eq!(span.name, "SELECT users");
    assert_eq!(
        attr(span, "db.system.name"),
        Some(opentelemetry::Value::String(dialect.system.into())),
    );
    assert_eq!(
        attr(span, "db.operation.name"),
        Some(opentelemetry::Value::String("SELECT".into())),
    );
    assert_eq!(
        attr(span, "db.collection.name"),
        Some(opentelemetry::Value::String("users".into())),
    );
}

/// Assert that the exporter contains exactly one span and that it matches the standard
/// annotation shape. Used to collapse the recurring trailing assertion block at the end
/// of every annotation-style test (including the `sqlx::query!()` macro tests, whose
/// bodies must remain backend-specific but whose assertions can reuse this helper).
pub fn assert_one_annotated_span(tel: &TestTelemetry, dialect: &Dialect) {
    let spans = tel.spans();
    assert_eq!(
        spans.len(),
        1,
        "expected exactly one span, got {}",
        spans.len()
    );
    assert_annotated_span(&spans[0], dialect);
}

/// Compare two single-span exporter snapshots and assert that the executor-side and
/// query-side annotation surfaces emitted byte-identical span data on the dimensions
/// users observe (name, kind, and the full annotation/connection attribute set). The
/// `case` argument is included in failure messages so the cause is obvious when one of
/// several builder-family pairs in a parity test fails.
pub fn assert_span_parity(case: &str, exec_spans: &[SpanData], query_spans: &[SpanData]) {
    assert_eq!(
        exec_spans.len(),
        1,
        "{case}: executor-side emitted {} spans, expected 1",
        exec_spans.len()
    );
    assert_eq!(
        query_spans.len(),
        1,
        "{case}: query-side emitted {} spans, expected 1",
        query_spans.len()
    );
    let exec = &exec_spans[0];
    let query = &query_spans[0];

    assert_eq!(exec.name, query.name, "{case}: span name differs");
    assert_eq!(exec.span_kind, query.span_kind, "{case}: span kind differs");
    for key in &[
        "db.system.name",
        "db.operation.name",
        "db.collection.name",
        "db.query.text",
        "db.query.summary",
        "db.namespace",
        "db.stored_procedure.name",
        "server.address",
        "server.port",
        "db.response.affected_rows",
        "db.response.returned_rows",
    ] {
        assert_eq!(
            attr(exec, key),
            attr(query, key),
            "{case}: attribute `{key}` differs across executor-side vs query-side",
        );
    }
}

// ---------------------------------------------------------------------------
// Parameterised test bodies (macro_rules)
// ---------------------------------------------------------------------------
//
// Each macro takes a pool factory expression and a dialect constant and expands to
// the full test body. Backend wrappers invoke with their factory + dialect:
//
//     #[tokio::test]
//     #[serial]
//     async fn execute_creates_span_via_pool() {
//         test_execute_creates_span_via_pool!(test_pool().await, common::SQLITE_DIALECT);
//     }

/// Bound-chain proof: exercises `&Pool<DB>: Executor` (plain), `Annotated<'_,
/// Pool<DB>>: Executor` (`with_annotations`), and the same via the `with_operation`
/// shorthand. Each `pool.execute(...)` runs against a freshly created table, so the
/// macro is safe to invoke against shared postgres / mysql containers.
#[macro_export]
macro_rules! test_execute_creates_span_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;
        $crate::fresh_table!(
            &pool,
            "exec_pool_test",
            &format!("id {}", $dialect.id_pk_column)
        );
        tel.reset();

        (&pool)
            .execute("INSERT INTO exec_pool_test (id) VALUES (1)")
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());
        assert!($crate::common::attr(&spans[0], "db.response.affected_rows").is_some());

        pool.with_annotations($crate::common::test_annotations())
            .execute("INSERT INTO exec_pool_test (id) VALUES (2)")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        pool.with_operation("SELECT", "users")
            .execute("INSERT INTO exec_pool_test (id) VALUES (3)")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Counterpart to `test_execute_creates_span_via_pool!` for `&mut PoolConnection<DB>`.
/// Acquires a connection from the pool, then exercises plain / annotated / shorthand
/// executes against a freshly created table.
#[macro_export]
macro_rules! test_execute_creates_span_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;
        $crate::fresh_table!(
            &pool,
            "exec_conn_test",
            &format!("id {}", $dialect.id_pk_column)
        );
        tel.reset();

        let mut conn = pool.acquire().await.unwrap();
        (&mut conn)
            .execute("INSERT INTO exec_conn_test (id) VALUES (1)")
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());
        assert!($crate::common::attr(&spans[0], "db.response.affected_rows").is_some());

        conn.with_annotations($crate::common::test_annotations())
            .execute("INSERT INTO exec_conn_test (id) VALUES (2)")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        conn.with_operation("SELECT", "users")
            .execute("INSERT INTO exec_conn_test (id) VALUES (3)")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Counterpart to `test_execute_creates_span_via_pool!` for `&mut Transaction<'_, DB>`.
/// Begins a transaction, runs three executes, and commits before asserting on the
/// collected spans.
#[macro_export]
macro_rules! test_execute_creates_span_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;
        $crate::fresh_table!(
            &pool,
            "exec_tx_test",
            &format!("id {}", $dialect.id_pk_column)
        );
        tel.reset();

        let mut tx = pool.begin().await.unwrap();
        (&mut tx)
            .execute("INSERT INTO exec_tx_test (id) VALUES (1)")
            .await
            .unwrap();

        tx.with_annotations($crate::common::test_annotations())
            .execute("INSERT INTO exec_tx_test (id) VALUES (2)")
            .await
            .unwrap();

        tx.with_operation("SELECT", "users")
            .execute("INSERT INTO exec_tx_test (id) VALUES (3)")
            .await
            .unwrap();

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());
        assert!($crate::common::attr(&spans[0], "db.response.affected_rows").is_some());
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `execute` against invalid SQL records an error span. Exercises plain, annotated, and
/// shorthand annotation paths.
#[macro_export]
macro_rules! test_execute_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result = (&pool).execute("INVALID SQL GIBBERISH").await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        let result = pool
            .with_annotations($crate::common::test_annotations())
            .execute("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let result = pool
            .with_operation("SELECT", "users")
            .execute("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `execute_many` over a multi-statement query yields one span per stream consumption.
/// Exercises plain, annotated, and shorthand paths against the wrapped pool.
#[macro_export]
macro_rules! test_execute_many_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = (&pool).execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );

        let mut stream = pool
            .with_annotations($crate::common::test_annotations())
            .execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        let mut stream = pool
            .with_operation("SELECT", "users")
            .execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `execute_many` against `&mut PoolConnection<DB>`. Same shape as the pool variant
/// but acquires a connection first.
#[macro_export]
macro_rules! test_execute_many_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let mut stream = (&mut conn).execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );

        let mut stream = conn
            .with_annotations($crate::common::test_annotations())
            .execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        let mut stream = conn
            .with_operation("SELECT", "users")
            .execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `execute_many` against `&mut Transaction<'_, DB>`. Asserts on all three spans
/// after commit.
#[macro_export]
macro_rules! test_execute_many_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let mut stream = (&mut tx).execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let mut stream = tx
            .with_annotations($crate::common::test_annotations())
            .execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let mut stream = tx
            .with_operation("SELECT", "users")
            .execute_many("SELECT 1; SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `execute_many` against invalid SQL records an error span on the streaming path.
#[macro_export]
macro_rules! test_execute_many_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = (&pool).execute_many("INVALID SQL GIBBERISH");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );

        let mut stream = pool
            .with_annotations($crate::common::test_annotations())
            .execute_many("INVALID SQL GIBBERISH");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let mut stream = pool
            .with_operation("SELECT", "users")
            .execute_many("INVALID SQL GIBBERISH");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `fetch` against the wrapped pool. Streams 2 rows, then exercises annotated and
/// shorthand variants.
#[macro_export]
macro_rules! test_fetch_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = (&pool).fetch("SELECT 1 UNION ALL SELECT 2");
        let mut count = 0u64;
        while stream.next().await.is_some() {
            count += 1;
        }
        assert_eq!(count, 2);
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );

        let mut stream = pool
            .with_annotations($crate::common::test_annotations())
            .fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        let mut stream = pool
            .with_operation("SELECT", "users")
            .fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_fetch_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let mut stream = (&mut conn).fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );

        let mut stream = conn
            .with_annotations($crate::common::test_annotations())
            .fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        let mut stream = conn
            .with_operation("SELECT", "users")
            .fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_fetch_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let mut stream = (&mut tx).fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let mut stream = tx
            .with_annotations($crate::common::test_annotations())
            .fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let mut stream = tx
            .with_operation("SELECT", "users")
            .fetch("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Verifies that dropping a `fetch` stream after consuming a single row still
/// finalises and exports the span (with `returned_rows` reflecting the partial read).
#[macro_export]
macro_rules! test_fetch_stream_dropped_early_still_records_span {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        {
            let mut stream = (&pool).fetch("SELECT 1 UNION ALL SELECT 2");
            let _ = stream.next().await;
        }

        let spans = tel.spans();
        assert_eq!(
            spans.len(),
            1,
            "span should be recorded even when stream is dropped early"
        );
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );

        $crate::common::assert_metric_for_system(&tel, $dialect.system);
    }};
}

/// `fetch` against invalid SQL records an error span on the streaming path.
#[macro_export]
macro_rules! test_fetch_stream_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = (&pool).fetch("INVALID SQL");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );

        let mut stream = pool
            .with_annotations($crate::common::test_annotations())
            .fetch("INVALID SQL");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let mut stream = pool.with_operation("SELECT", "users").fetch("INVALID SQL");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `fetch_many` against the wrapped pool. Returns rows + result rows on a stream.
#[macro_export]
macro_rules! test_fetch_many_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = (&pool).fetch_many("SELECT 1 UNION ALL SELECT 2");
        let mut rows = 0u64;
        let mut results = 0u64;
        while let Some(item) = stream.next().await {
            match item.unwrap() {
                sqlx::Either::Left(_) => results += 1,
                sqlx::Either::Right(_) => rows += 1,
            }
        }
        drop(stream);

        assert_eq!(rows, 2);
        assert!(results >= 1, "should have at least one QueryResult");

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );

        let mut stream = pool
            .with_annotations($crate::common::test_annotations())
            .fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        let mut stream = pool
            .with_operation("SELECT", "users")
            .fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_many` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_fetch_many_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let mut stream = (&mut conn).fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );

        let mut stream = conn
            .with_annotations($crate::common::test_annotations())
            .fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        let mut stream = conn
            .with_operation("SELECT", "users")
            .fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_many` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_fetch_many_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let mut stream = (&mut tx).fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let mut stream = tx
            .with_annotations($crate::common::test_annotations())
            .fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        let mut stream = tx
            .with_operation("SELECT", "users")
            .fetch_many("SELECT 1 UNION ALL SELECT 2");
        while stream.next().await.is_some() {}
        drop(stream);

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Verifies that dropping a `fetch_many` stream after consuming a single row still
/// finalises and exports the span.
#[macro_export]
macro_rules! test_fetch_many_dropped_early_still_records_span {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        {
            let mut stream = (&pool).fetch_many("SELECT 1 UNION ALL SELECT 2");
            let _ = stream.next().await;
        }

        let spans = tel.spans();
        assert_eq!(
            spans.len(),
            1,
            "span should be recorded even when stream is dropped early"
        );
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );

        $crate::common::assert_metric_for_system(&tel, $dialect.system);
    }};
}

/// `fetch_many` against invalid SQL records an error span on the streaming path.
#[macro_export]
macro_rules! test_fetch_many_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = (&pool).fetch_many("INVALID SQL GIBBERISH");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );

        let mut stream = pool
            .with_annotations($crate::common::test_annotations())
            .fetch_many("INVALID SQL GIBBERISH");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let mut stream = pool
            .with_operation("SELECT", "users")
            .fetch_many("INVALID SQL GIBBERISH");
        let result = stream.next().await;
        assert!(result.is_some_and(|r| r.is_err()));
        drop(stream);
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `fetch_all` against the wrapped pool. Returns 3 rows; exercises plain, annotated,
/// and shorthand variants.
#[macro_export]
macro_rules! test_fetch_all_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let rows = (&pool)
            .fetch_all("SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3")
            .await
            .unwrap();
        assert_eq!(rows.len(), 3);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(3))
        );

        pool.with_annotations($crate::common::test_annotations())
            .fetch_all("SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        pool.with_operation("SELECT", "users")
            .fetch_all("SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_all` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_fetch_all_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let rows = (&mut conn)
            .fetch_all("SELECT 1 UNION ALL SELECT 2")
            .await
            .unwrap();
        assert_eq!(rows.len(), 2);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );

        conn.with_annotations($crate::common::test_annotations())
            .fetch_all("SELECT 1 UNION ALL SELECT 2")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        conn.with_operation("SELECT", "users")
            .fetch_all("SELECT 1 UNION ALL SELECT 2")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_all` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_fetch_all_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let rows = (&mut tx)
            .fetch_all("SELECT 1 UNION ALL SELECT 2")
            .await
            .unwrap();
        assert_eq!(rows.len(), 2);

        tx.with_annotations($crate::common::test_annotations())
            .fetch_all("SELECT 1 UNION ALL SELECT 2")
            .await
            .unwrap();

        tx.with_operation("SELECT", "users")
            .fetch_all("SELECT 1 UNION ALL SELECT 2")
            .await
            .unwrap();

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_all` against invalid SQL records an error span.
#[macro_export]
macro_rules! test_fetch_all_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result = (&pool).fetch_all("INVALID SQL GIBBERISH").await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        let result = pool
            .with_annotations($crate::common::test_annotations())
            .fetch_all("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let result = pool
            .with_operation("SELECT", "users")
            .fetch_all("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `fetch_one` against the wrapped pool. Returns 1 row.
#[macro_export]
macro_rules! test_fetch_one_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let _row = (&pool).fetch_one("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );

        pool.with_annotations($crate::common::test_annotations())
            .fetch_one("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        pool.with_operation("SELECT", "users")
            .fetch_one("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_one` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_fetch_one_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let _row = (&mut conn).fetch_one("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );

        conn.with_annotations($crate::common::test_annotations())
            .fetch_one("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        conn.with_operation("SELECT", "users")
            .fetch_one("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_one` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_fetch_one_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let _row = (&mut tx).fetch_one("SELECT 1").await.unwrap();

        tx.with_annotations($crate::common::test_annotations())
            .fetch_one("SELECT 1")
            .await
            .unwrap();

        tx.with_operation("SELECT", "users")
            .fetch_one("SELECT 1")
            .await
            .unwrap();

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_one` against invalid SQL records an error span.
#[macro_export]
macro_rules! test_fetch_one_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result = (&pool).fetch_one("INVALID SQL GIBBERISH").await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        let result = pool
            .with_annotations($crate::common::test_annotations())
            .fetch_one("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let result = pool
            .with_operation("SELECT", "users")
            .fetch_one("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `fetch_optional` against the wrapped pool when the query returns one row.
#[macro_export]
macro_rules! test_fetch_optional_records_one_row {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result = (&pool).fetch_optional("SELECT 1").await.unwrap();
        assert!(result.is_some());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );

        pool.with_annotations($crate::common::test_annotations())
            .fetch_optional("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        pool.with_operation("SELECT", "users")
            .fetch_optional("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_optional` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_fetch_optional_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let result = (&mut conn).fetch_optional("SELECT 1").await.unwrap();
        assert!(result.is_some());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );

        conn.with_annotations($crate::common::test_annotations())
            .fetch_optional("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        conn.with_operation("SELECT", "users")
            .fetch_optional("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_optional` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_fetch_optional_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let result = (&mut tx).fetch_optional("SELECT 1").await.unwrap();
        assert!(result.is_some());

        tx.with_annotations($crate::common::test_annotations())
            .fetch_optional("SELECT 1")
            .await
            .unwrap();

        tx.with_operation("SELECT", "users")
            .fetch_optional("SELECT 1")
            .await
            .unwrap();

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `fetch_optional` against invalid SQL records an error span.
#[macro_export]
macro_rules! test_fetch_optional_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result = (&pool).fetch_optional("INVALID SQL GIBBERISH").await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            None
        );

        let result = pool
            .with_annotations($crate::common::test_annotations())
            .fetch_optional("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let result = pool
            .with_operation("SELECT", "users")
            .fetch_optional("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

// ---------------------------------------------------------------------------
// prepare / prepare_with / describe
// ---------------------------------------------------------------------------

/// `prepare` against the wrapped pool. No rows returned; just verifies the span shape.
#[macro_export]
macro_rules! test_prepare_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let _stmt = (&pool).prepare("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        pool.with_annotations($crate::common::test_annotations())
            .prepare("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        pool.with_operation("SELECT", "users")
            .prepare("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `prepare` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_prepare_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let _stmt = (&mut conn).prepare("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        conn.with_annotations($crate::common::test_annotations())
            .prepare("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        conn.with_operation("SELECT", "users")
            .prepare("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `prepare` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_prepare_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let _stmt = (&mut tx).prepare("SELECT 1").await.unwrap();

        tx.with_annotations($crate::common::test_annotations())
            .prepare("SELECT 1")
            .await
            .unwrap();

        tx.with_operation("SELECT", "users")
            .prepare("SELECT 1")
            .await
            .unwrap();

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `prepare` against invalid SQL records an error span.
#[macro_export]
macro_rules! test_prepare_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let result = (&mut conn).prepare("INVALID SQL GIBBERISH").await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        let result = conn
            .with_annotations($crate::common::test_annotations())
            .prepare("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let result = conn
            .with_operation("SELECT", "users")
            .prepare("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `prepare_with` against the wrapped pool.
#[macro_export]
macro_rules! test_prepare_with_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let _stmt = (&pool)
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        pool.with_annotations($crate::common::test_annotations())
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        pool.with_operation("SELECT", "users")
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `prepare_with` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_prepare_with_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let _stmt = (&mut conn)
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        conn.with_annotations($crate::common::test_annotations())
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        conn.with_operation("SELECT", "users")
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `prepare_with` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_prepare_with_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let _stmt = (&mut tx)
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();

        tx.with_annotations($crate::common::test_annotations())
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();

        tx.with_operation("SELECT", "users")
            .prepare_with($dialect.prepare_with_select_sql, &[])
            .await
            .unwrap();

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `prepare_with` against invalid SQL records an error span.
#[macro_export]
macro_rules! test_prepare_with_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let result = (&mut conn).prepare_with("INVALID SQL GIBBERISH", &[]).await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        let result = conn
            .with_annotations($crate::common::test_annotations())
            .prepare_with("INVALID SQL GIBBERISH", &[])
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let result = conn
            .with_operation("SELECT", "users")
            .prepare_with("INVALID SQL GIBBERISH", &[])
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `describe` against the wrapped pool.
#[macro_export]
macro_rules! test_describe_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let _desc = (&pool).describe("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        pool.with_annotations($crate::common::test_annotations())
            .describe("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        pool.with_operation("SELECT", "users")
            .describe("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `describe` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_describe_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let _desc = (&mut conn).describe("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        conn.with_annotations($crate::common::test_annotations())
            .describe("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        conn.with_operation("SELECT", "users")
            .describe("SELECT 1")
            .await
            .unwrap();
        $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `describe` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_describe_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let _desc = (&mut tx).describe("SELECT 1").await.unwrap();

        tx.with_annotations($crate::common::test_annotations())
            .describe("SELECT 1")
            .await
            .unwrap();

        tx.with_operation("SELECT", "users")
            .describe("SELECT 1")
            .await
            .unwrap();

        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 3);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());
        $crate::common::assert_annotated_span(&spans[1], &$dialect);
        $crate::common::assert_annotated_span(&spans[2], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `describe` against invalid SQL records an error span.
#[macro_export]
macro_rules! test_describe_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let result = (&mut conn).describe("INVALID SQL GIBBERISH").await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        $crate::common::assert_error_span(&spans[0]);
        assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none());

        let result = conn
            .with_annotations($crate::common::test_annotations())
            .describe("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        let result = conn
            .with_operation("SELECT", "users")
            .describe("INVALID SQL GIBBERISH")
            .await;
        assert!(result.is_err());
        let last = tel.spans().last().unwrap().clone();
        $crate::common::assert_annotated_span(&last, &$dialect);
        $crate::common::assert_error_span(&last);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

// ---------------------------------------------------------------------------
// Misc: metrics, annotations
// ---------------------------------------------------------------------------

/// All four annotation fields set together must surface on the `db.client.operation.duration`
/// histogram data point. Per-method macros only exercise the standard
/// `test_annotations()` shape (`db.operation.name = "SELECT"`,
/// `db.collection.name = "users"`), so this macro pins the integration-level guarantee
/// for `db.query.summary` and `db.stored_procedure.name` propagation.
#[macro_export]
macro_rules! test_operation_duration_metric_carries_full_annotations {
    ($pool_factory:expr, $dialect:expr) => {{
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        pool.with_annotations(
            sqlx_otel::QueryAnnotations::new()
                .operation("SELECT")
                .collection("users")
                .query_summary("users by id")
                .stored_procedure("sp_get_users"),
        )
        .fetch_one("SELECT 1")
        .await
        .unwrap();

        $crate::common::assert_metric_data_point(
            &tel,
            &[
                ("db.system.name", $dialect.system),
                ("db.operation.name", "SELECT"),
                ("db.collection.name", "users"),
                ("db.query.summary", "users by id"),
                ("db.stored_procedure.name", "sp_get_users"),
            ],
        );
    }};
}

/// Targeted SQLSTATE assertion: on `sqlx::Error::Database`, the backend status code
/// surfaces on the histogram as `db.response.status_code`. The expected code is backend-
/// specific: `SQLite` extended result code `1` (`SQLITE_ERROR`), Postgres SQLSTATE
/// `42P01`, `MySQL` SQLSTATE `42S02` – each backend's `tests/{sqlite,postgres,mysql}.rs`
/// passes the value it expects. Per-method `*_records_error` macros already assert the
/// generic `error.type` propagation; this macro pins the SQLSTATE shape that varies per
/// backend.
#[macro_export]
macro_rules! test_operation_duration_metric_carries_sqlstate {
    ($pool_factory:expr, $expected_code:expr) => {{
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result = pool
            .with_operation("SELECT", "nonexistent_table_xyz")
            .fetch_one("SELECT * FROM nonexistent_table_xyz")
            .await;
        assert!(result.is_err(), "expected fetch_one to fail");

        let resource_metrics = tel.metrics();
        let dp = $crate::common::find_duration_data_point(&resource_metrics)
            .expect("db.client.operation.duration data point missing");

        assert_eq!(
            $crate::common::metric_attr(&dp, "db.response.status_code"),
            Some(opentelemetry::Value::String($expected_code.into())),
            "metric must carry db.response.status_code for sqlx::Error::Database",
        );
    }};
}

/// All four annotation fields populated together; summary drives the span name.
#[macro_export]
macro_rules! test_annotation_all_four_fields {
    ($pool_factory:expr, $dialect:expr) => {{
        let _ = $dialect; // dialect.system is checked indirectly via assert_common_span_attributes when present
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        pool.with_annotations(
            sqlx_otel::QueryAnnotations::new()
                .operation("SELECT")
                .collection("users")
                .query_summary("users by id")
                .stored_procedure("sp_get_users"),
        )
        .fetch_all("SELECT 1")
        .await
        .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(spans[0].name, "users by id");
        assert_eq!(
            $crate::common::attr(&spans[0], "db.operation.name"),
            Some(opentelemetry::Value::String("SELECT".into())),
        );
        assert_eq!(
            $crate::common::attr(&spans[0], "db.collection.name"),
            Some(opentelemetry::Value::String("users".into())),
        );
        assert_eq!(
            $crate::common::attr(&spans[0], "db.query.summary"),
            Some(opentelemetry::Value::String("users by id".into())),
        );
        assert_eq!(
            $crate::common::attr(&spans[0], "db.stored_procedure.name"),
            Some(opentelemetry::Value::String("sp_get_users".into())),
        );
    }};
}

/// `db.query.summary` overrides the span name independently of `db.operation.name` and
/// `db.collection.name`, but does not suppress those attributes.
#[macro_export]
macro_rules! test_query_summary_drives_span_name {
    ($pool_factory:expr, $dialect:expr) => {{
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        pool.with_annotations(
            sqlx_otel::QueryAnnotations::new()
                .operation("SELECT")
                .collection("users")
                .query_summary("users by tenant"),
        )
        .fetch_all("SELECT 1")
        .await
        .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(spans[0].name, "users by tenant");
        assert_eq!(
            $crate::common::attr(&spans[0], "db.query.summary"),
            Some(opentelemetry::Value::String("users by tenant".into())),
        );
        assert_eq!(
            $crate::common::attr(&spans[0], "db.operation.name"),
            Some(opentelemetry::Value::String("SELECT".into())),
        );
        assert_eq!(
            $crate::common::attr(&spans[0], "db.collection.name"),
            Some(opentelemetry::Value::String("users".into())),
        );
    }};
}

// ---------------------------------------------------------------------------
// query-side annotations: sqlx::query(...).with_annotations(...).<method>(executor)
// ---------------------------------------------------------------------------

/// `sqlx::query(...).with_annotations(...).execute_many(&pool)`.
#[macro_export]
macro_rules! test_query_execute_many_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        #[allow(deprecated)]
        let mut stream = sqlx::query("SELECT 1; SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .execute_many(&pool)
            .await;
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query(...).with_annotations(...).fetch(&pool)`.
#[macro_export]
macro_rules! test_query_fetch_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = sqlx::query("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query(...).with_annotations(...).fetch_many(&pool)`.
#[macro_export]
macro_rules! test_query_fetch_many_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        #[allow(deprecated)]
        let mut stream = sqlx::query("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch_many(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query(...).with_annotations(...).fetch_all(&pool)`.
#[macro_export]
macro_rules! test_query_fetch_all_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let rows = sqlx::query("SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3")
            .with_annotations($crate::common::test_annotations())
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(rows.len(), 3);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(3))
        );

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query(...).with_annotations(...).fetch_one(&pool)`.
#[macro_export]
macro_rules! test_query_fetch_one_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let _row = sqlx::query("SELECT 1")
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(1))
        );

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query("INVALID SQL").with_annotations(...).execute(&pool)` records an error span.
#[macro_export]
macro_rules! test_query_execute_with_annotations_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result = sqlx::query("INVALID SQL GIBBERISH")
            .with_annotations($crate::common::test_annotations())
            .execute(&pool)
            .await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        $crate::common::assert_error_span(&spans[0]);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_as(...).with_annotations(...).fetch(&pool)`.
#[macro_export]
macro_rules! test_query_as_fetch_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = sqlx::query_as::<_, (i32,)>("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_as(...).with_annotations(...).fetch_many(&pool)`.
#[macro_export]
macro_rules! test_query_as_fetch_many_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        #[allow(deprecated)]
        let mut stream = sqlx::query_as::<_, (i32,)>("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch_many(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_as(...).with_annotations(...).fetch_all(&pool)`.
#[macro_export]
macro_rules! test_query_as_fetch_all_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let rows: Vec<(i32,)> = sqlx::query_as("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(rows.len(), 2);

        let pool_clone = pool.clone();
        let rows: Vec<(i32,)> = tokio::spawn(async move {
            sqlx::query_as("SELECT 1 UNION ALL SELECT 2")
                .with_annotations($crate::common::test_annotations())
                .fetch_all(&pool_clone)
                .await
                .unwrap()
        })
        .await
        .unwrap();
        assert_eq!(rows.len(), 2);

        let spans = tel.spans();
        assert_eq!(spans.len(), 2);
        for span in &spans {
            $crate::common::assert_annotated_span(span, &$dialect);
        }

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_as(...).with_annotations(...).fetch_one(&pool)`. Runs inline and inside
/// `tokio::spawn` to exercise the `Send`-required path.
#[macro_export]
macro_rules! test_query_as_fetch_one_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let row: (i32,) = sqlx::query_as("SELECT 7")
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(row.0, 7);

        let pool_clone = pool.clone();
        let row: (i32,) = tokio::spawn(async move {
            sqlx::query_as("SELECT 7")
                .with_annotations($crate::common::test_annotations())
                .fetch_one(&pool_clone)
                .await
                .unwrap()
        })
        .await
        .unwrap();
        assert_eq!(row.0, 7);

        let spans = tel.spans();
        assert_eq!(spans.len(), 2);
        for span in &spans {
            $crate::common::assert_annotated_span(span, &$dialect);
        }

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_as(...).with_annotations(...).fetch_optional(&pool)` returning none.
/// Runs inline and inside `tokio::spawn` to exercise the `Send`-required path.
#[macro_export]
macro_rules! test_query_as_fetch_optional_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let row: Option<(i32,)> = sqlx::query_as("SELECT 1 WHERE 1 = 0")
            .with_annotations($crate::common::test_annotations())
            .fetch_optional(&pool)
            .await
            .unwrap();
        assert!(row.is_none());

        let pool_clone = pool.clone();
        let row: Option<(i32,)> = tokio::spawn(async move {
            sqlx::query_as("SELECT 1 WHERE 1 = 0")
                .with_annotations($crate::common::test_annotations())
                .fetch_optional(&pool_clone)
                .await
                .unwrap()
        })
        .await
        .unwrap();
        assert!(row.is_none());

        let spans = tel.spans();
        assert_eq!(spans.len(), 2);
        for span in &spans {
            $crate::common::assert_annotated_span(span, &$dialect);
        }

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_as("INVALID SQL").with_annotations(...).fetch_one(&pool)` records error.
#[macro_export]
macro_rules! test_query_as_fetch_one_with_annotations_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result: Result<(i32,), _> = sqlx::query_as("INVALID SQL")
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        $crate::common::assert_error_span(&spans[0]);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_scalar(...).with_annotations(...).fetch(&pool)`.
#[macro_export]
macro_rules! test_query_scalar_fetch_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = sqlx::query_scalar::<_, i32>("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_scalar(...).with_annotations(...).fetch_many(&pool)`.
#[macro_export]
macro_rules! test_query_scalar_fetch_many_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        #[allow(deprecated)]
        let mut stream = sqlx::query_scalar::<_, i32>("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch_many(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_scalar(...).with_annotations(...).fetch_all(&pool)`.
#[macro_export]
macro_rules! test_query_scalar_fetch_all_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let rows: Vec<i32> = sqlx::query_scalar("SELECT 1 UNION ALL SELECT 2")
            .with_annotations($crate::common::test_annotations())
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(rows, vec![1, 2]);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_scalar(...).with_annotations(...).fetch_one(&pool)`.
#[macro_export]
macro_rules! test_query_scalar_fetch_one_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query_scalar("SELECT 42")
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 42);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `sqlx::query_scalar(...).with_annotations(...).fetch_optional(&pool)` returning none.
#[macro_export]
macro_rules! test_query_scalar_fetch_optional_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: Option<i32> = sqlx::query_scalar("SELECT 1 WHERE 1 = 0")
            .with_annotations($crate::common::test_annotations())
            .fetch_optional(&pool)
            .await
            .unwrap();
        assert!(value.is_none());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

// ---------------------------------------------------------------------------
// query-side annotations: Map (Query::map / Query::try_map)
// ---------------------------------------------------------------------------
//
// The closure parameter type for `.map(|row| ...)` is inferred from the surrounding
// `Query<DB>` chain, so we don't need to spell out per-backend `SqliteRow` / `PgRow` /
// `MySqlRow`.

/// `Query::with_annotations` before `bind`/`map`. Position-1 in the builder pipeline.
#[macro_export]
macro_rules! test_query_map_position_1_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query("SELECT 7")
            .with_annotations($crate::common::test_annotations())
            .map(|row: Row| row.get::<i32, _>(0))
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 7);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::with_annotations` after `bind`, before `map`.
#[macro_export]
macro_rules! test_query_map_position_2_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query("SELECT 11")
            .with_annotations($crate::common::test_annotations())
            .map(|row: Row| row.get::<i32, _>(0))
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 11);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::with_annotations` after `bind` and `map` – last in the pipeline.
#[macro_export]
macro_rules! test_query_map_position_3_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query("SELECT 13")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 13);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Same as `query_map_position_3` but with `try_map`.
#[macro_export]
macro_rules! test_query_try_map_position_3_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query("SELECT 17")
            .try_map(|row: Row| Ok(row.get::<i32, _>(0)))
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 17);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` then `with_annotations` then `fetch(&pool)`.
#[macro_export]
macro_rules! test_map_fetch_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut stream = sqlx::query("SELECT 1 UNION ALL SELECT 2")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(2))
        );

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` then `with_annotations` then `fetch_many(&pool)`.
#[macro_export]
macro_rules! test_map_fetch_many_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use futures::StreamExt as _;
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        #[allow(deprecated)]
        let mut stream = sqlx::query("SELECT 1 UNION ALL SELECT 2")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_many(&pool);
        while stream.next().await.is_some() {}
        drop(stream);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` then `with_annotations` then `fetch_all(&pool)`.
#[macro_export]
macro_rules! test_map_fetch_all_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let rows: Vec<i32> = sqlx::query("SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(rows, vec![1, 2, 3]);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` then `with_annotations` then `fetch_one(&pool)`.
#[macro_export]
macro_rules! test_map_fetch_one_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query("SELECT 19")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 19);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` then `with_annotations` then `fetch_optional(&pool)`.
#[macro_export]
macro_rules! test_map_fetch_optional_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: Option<i32> = sqlx::query("SELECT 1 WHERE 1 = 0")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_optional(&pool)
            .await
            .unwrap();
        assert!(value.is_none());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Composing two `map` calls, with annotations between them.
#[macro_export]
macro_rules! test_map_compose_after_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query("SELECT 5")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .map(|n| n * 2)
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 10);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Composing `map` then `try_map`, with annotations between.
#[macro_export]
macro_rules! test_map_try_map_compose_after_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let value: i32 = sqlx::query("SELECT 6")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .try_map(|n: i32| Ok::<_, sqlx::Error>(n + 100))
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(value, 106);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` against `&mut PoolConnection<DB>`.
#[macro_export]
macro_rules! test_query_map_with_annotations_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        let value: i32 = sqlx::query("SELECT 23")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&mut conn)
            .await
            .unwrap();
        assert_eq!(value, 23);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` against `&mut Transaction<'_, DB>`.
#[macro_export]
macro_rules! test_query_map_with_annotations_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        let value: i32 = sqlx::query("SELECT 29")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&mut tx)
            .await
            .unwrap();
        assert_eq!(value, 29);
        tx.commit().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `Query::map` against invalid SQL records an error span.
#[macro_export]
macro_rules! test_query_map_with_annotations_records_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result: Result<i32, _> = sqlx::query("INVALID SQL")
            .map(|row: Row| row.get::<i32, _>(0))
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        $crate::common::assert_error_span(&spans[0]);

        $crate::common::assert_annotated_error_metric(&tel, &$dialect);
    }};
}

/// `try_map` returning a mapper-side error: the database round-trip succeeds, the span
/// reports success, but the user-visible `Result` carries the mapper's error.
#[macro_export]
macro_rules! test_query_try_map_with_annotations_propagates_mapper_error {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let result: Result<i64, _> = sqlx::query("SELECT 1")
            .try_map(|_row: Row| {
                Err::<i64, _>(sqlx::Error::Decode(
                    "intentional decode failure".to_string().into(),
                ))
            })
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await;
        assert!(result.is_err());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

// ---------------------------------------------------------------------------
// PoolBuilder with_* methods
// ---------------------------------------------------------------------------
//
// These macros accept a *raw* pool factory (not the wrapped `Pool<DB>`) so the test
// body can configure `PoolBuilder` itself with the override under test.

/// `PoolBuilder::with_database` overrides the inferred `db.namespace`.
#[macro_export]
macro_rules! test_builder_with_database_overrides_namespace {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_database("custom_db")
            .build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.namespace"),
            Some(opentelemetry::Value::String("custom_db".into()))
        );
    }};
}

/// `PoolBuilder::with_host` overrides the inferred `server.address`.
#[macro_export]
macro_rules! test_builder_with_host_overrides_server_address {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_host("custom-host")
            .build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "server.address"),
            Some(opentelemetry::Value::String("custom-host".into()))
        );
    }};
}

/// `PoolBuilder::with_port` overrides the inferred `server.port`.
#[macro_export]
macro_rules! test_builder_with_port_overrides_server_port {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw).with_port(9999).build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "server.port"),
            Some(opentelemetry::Value::I64(9999))
        );
    }};
}

/// `PoolBuilder::with_network_peer_address` populates `network.peer.address`.
#[macro_export]
macro_rules! test_builder_with_network_peer_address {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_network_peer_address("10.0.0.5")
            .build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "network.peer.address"),
            Some(opentelemetry::Value::String("10.0.0.5".into()))
        );
    }};
}

/// `PoolBuilder::with_pool_name` propagates `db.client.connection.pool.name` onto every
/// span and per-operation metric so dashboards can correlate pool-level signals (the
/// `db.client.connection.*` family) with query-level latency.
#[macro_export]
macro_rules! test_builder_with_pool_name_propagates_to_span_and_metric {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_pool_name("primary-rw")
            .build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.client.connection.pool.name"),
            Some(opentelemetry::Value::String("primary-rw".into())),
            "span must carry db.client.connection.pool.name set via with_pool_name",
        );

        let resource_metrics = tel.metrics();
        let dp = $crate::common::find_duration_data_point(&resource_metrics)
            .expect("db.client.operation.duration data point missing");
        assert_eq!(
            $crate::common::metric_attr(&dp, "db.client.connection.pool.name"),
            Some(opentelemetry::Value::String("primary-rw".into())),
            "metric must carry db.client.connection.pool.name set via with_pool_name",
        );
    }};
}

/// Default `network.protocol.name` is the backend's wire protocol (Postgres / `MySQL`)
/// or absent (`SQLite`). The expected value is supplied per backend because `Dialect`
/// does not currently distinguish wire-protocol-bearing backends from embedded ones.
#[macro_export]
macro_rules! test_builder_default_network_protocol_name {
    ($raw_pool_factory:expr, $expected:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw).build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        let actual_span = $crate::common::attr(&spans[0], "network.protocol.name");
        let expected: Option<&str> = $expected;
        assert_eq!(
            actual_span,
            expected.map(|s| opentelemetry::Value::String(s.to_owned().into())),
            "span network.protocol.name must match the backend default",
        );

        let resource_metrics = tel.metrics();
        let dp = $crate::common::find_duration_data_point(&resource_metrics)
            .expect("db.client.operation.duration data point missing");
        let actual_metric = $crate::common::metric_attr(&dp, "network.protocol.name");
        assert_eq!(
            actual_metric,
            expected.map(|s| opentelemetry::Value::String(s.to_owned().into())),
            "metric network.protocol.name must match the backend default",
        );
    }};
}

/// `PoolBuilder::with_network_protocol_name` overrides the backend default and the value
/// surfaces on both spans and per-operation metrics.
#[macro_export]
macro_rules! test_builder_with_network_protocol_name_overrides {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_network_protocol_name("custom-proto")
            .build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "network.protocol.name"),
            Some(opentelemetry::Value::String("custom-proto".into())),
            "with_network_protocol_name must override the default on the span",
        );

        let resource_metrics = tel.metrics();
        let dp = $crate::common::find_duration_data_point(&resource_metrics)
            .expect("db.client.operation.duration data point missing");
        assert_eq!(
            $crate::common::metric_attr(&dp, "network.protocol.name"),
            Some(opentelemetry::Value::String("custom-proto".into())),
            "with_network_protocol_name must override the default on the metric",
        );
    }};
}

/// `PoolBuilder::with_network_transport` propagates `network.transport` onto spans and
/// per-operation metrics. The wrapper does not infer transport from the connect string,
/// so the attribute reflects the deployment configuration the caller declared.
#[macro_export]
macro_rules! test_builder_with_network_transport {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_network_transport("tcp")
            .build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "network.transport"),
            Some(opentelemetry::Value::String("tcp".into())),
            "span must carry network.transport set via with_network_transport",
        );

        let resource_metrics = tel.metrics();
        let dp = $crate::common::find_duration_data_point(&resource_metrics)
            .expect("db.client.operation.duration data point missing");
        assert_eq!(
            $crate::common::metric_attr(&dp, "network.transport"),
            Some(opentelemetry::Value::String("tcp".into())),
            "metric must carry network.transport set via with_network_transport",
        );
    }};
}

/// `PoolBuilder::with_network_peer_port` populates `network.peer.port`.
#[macro_export]
macro_rules! test_builder_with_network_peer_port {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let _ = $dialect;
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_network_peer_port(5433)
            .build();

        let _ = (&pool).fetch_optional("SELECT 1").await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "network.peer.port"),
            Some(opentelemetry::Value::I64(5433))
        );
    }};
}

/// `Pool::close` and `Pool::is_closed` round-trip.
#[macro_export]
macro_rules! test_pool_close_and_is_closed {
    ($pool_factory:expr, $dialect:expr) => {{
        let _ = $dialect;
        let _tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        assert!(!pool.is_closed());
        pool.close().await;
        assert!(pool.is_closed());
    }};
}

/// `QueryTextMode::Off` suppresses the `db.query.text` attribute.
#[macro_export]
macro_rules! test_query_text_mode_off_suppresses_sql {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        let tel = $crate::common::TestTelemetry::install();
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_query_text_mode(sqlx_otel::QueryTextMode::Off)
            .build();

        let _: Option<(i32,)> = sqlx::query_as("SELECT 1")
            .fetch_optional(&pool)
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(spans[0].span_kind, opentelemetry::trace::SpanKind::Client);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.system.name"),
            Some(opentelemetry::Value::String($dialect.system.into()))
        );
        assert!($crate::common::attr(&spans[0], "db.namespace").is_some());
        assert!(
            $crate::common::attr(&spans[0], "db.query.text").is_none(),
            "db.query.text should not be present when QueryTextMode::Off"
        );

        $crate::common::assert_metric_for_system(&tel, $dialect.system);
    }};
}

// ---------------------------------------------------------------------------
// Dialect-portable test bodies
// ---------------------------------------------------------------------------

/// `execute` records the correct `db.response.affected_rows` for a sequence of
/// INSERT / upsert / UPDATE / DELETE statements. Uses the dialect's `upsert_sql`,
/// `upsert_affected_rows`, and `string_concat_update_sql` to handle backend-specific
/// upsert syntax and string-concat operators.
#[macro_export]
macro_rules! test_execute_records_affected_rows {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;
        $crate::fresh_table!(
            &pool,
            "affected_test",
            &format!("id {}, name {}", $dialect.id_pk_column, $dialect.text_column)
        );
        tel.reset();

        // --- Bulk insert via VALUES list ---
        (&pool)
            .execute(
                "INSERT INTO affected_test (id, name) VALUES (1, 'alice'), (2, 'bob'), (3, 'carol')",
            )
            .await
            .unwrap();
        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.affected_rows"),
            Some(opentelemetry::Value::I64(3)),
            "inserting 3 rows should affect 3 rows"
        );
        $crate::common::assert_metric_for_system(&tel, $dialect.system);
        $crate::common::assert_affected_rows_metric(&tel, $dialect.system);
        tel.reset();

        // --- Upsert (dialect-specific) ---
        (&pool).execute($dialect.upsert_sql).await.unwrap();
        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.affected_rows"),
            Some(opentelemetry::Value::I64($dialect.upsert_affected_rows)),
            "upsert affected_rows differs per backend"
        );
        $crate::common::assert_affected_rows_metric(&tel, $dialect.system);
        tel.reset();

        // --- Update multiple rows (dialect-specific concat) ---
        (&pool)
            .execute($dialect.string_concat_update_sql)
            .await
            .unwrap();
        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.affected_rows"),
            Some(opentelemetry::Value::I64(2)),
            "updating two rows should affect 2 rows"
        );
        $crate::common::assert_affected_rows_metric(&tel, $dialect.system);
        tel.reset();

        // --- Delete multiple rows ---
        (&pool)
            .execute("DELETE FROM affected_test WHERE id IN (1, 2, 3)")
            .await
            .unwrap();
        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.affected_rows"),
            Some(opentelemetry::Value::I64(3)),
            "deleting three rows should affect 3 rows"
        );
        $crate::common::assert_affected_rows_metric(&tel, $dialect.system);
        tel.reset();

        // --- Delete with no matching rows ---
        (&pool)
            .execute("DELETE FROM affected_test WHERE id = 999")
            .await
            .unwrap();
        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.affected_rows"),
            Some(opentelemetry::Value::I64(0)),
            "deleting non-existent rows should affect 0 rows"
        );
        $crate::common::assert_metric_for_system(&tel, $dialect.system);
        $crate::common::assert_affected_rows_metric(&tel, $dialect.system);
    }};
}

/// Transaction rollback emits a single CREATE TABLE span and discards the table.
/// Uses `fresh_table!` so the test is repeatable against the shared postgres / mysql
/// containers.
#[macro_export]
macro_rules! test_transaction_rollback {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let pool = $pool_factory;
        // Pre-clean any leftover table before installing telemetry, so the rollback test
        // sees only its own span.
        let drop_sql = "DROP TABLE IF EXISTS rollback_test";
        (&pool).execute(drop_sql).await.unwrap();

        let tel = $crate::common::TestTelemetry::install();

        let mut tx = pool.begin().await.unwrap();
        let create_sql = format!("CREATE TABLE rollback_test (id {})", $dialect.id_pk_column);
        (&mut tx).execute(create_sql.as_str()).await.unwrap();
        tx.rollback().await.unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
    }};
}

/// `QueryTextMode::Obfuscated` rewrites string and numeric literals in `db.query.text`.
/// The query under test is dialect-neutral (`SELECT 1, 'alice', 3.14`), so the only
/// dialect input is the raw pool factory used to build a custom-configured pool.
#[macro_export]
macro_rules! test_query_text_mode_obfuscated_replaces_literals {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        // This macro intentionally narrows scope to span-side `db.query.text` capture; it
        // does not assert on metrics because the pool is constructed before the test
        // telemetry is installed, so the pool's metric instruments bind to the no-op meter.
        // Per-method metric coverage lives in every other `test_<method>_*` macro.
        use sqlx::Executor as _;
        let _ = $dialect;
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_query_text_mode(sqlx_otel::QueryTextMode::Obfuscated)
            .build();

        let tel = $crate::common::TestTelemetry::install();
        let _row = (&pool)
            .fetch_optional("SELECT 1, 'alice', 3.14")
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.query.text"),
            Some(opentelemetry::Value::String("SELECT ?, ?, ?".into()))
        );
    }};
}

/// `QueryTextMode::Full` (default) emits `db.query.text` with inter-token whitespace
/// collapsed to a single space. The query is dialect-neutral (no placeholders, just a
/// constant SELECT split across lines), so the only dialect input is the raw pool factory
/// used to build a default-configured pool.
#[macro_export]
macro_rules! test_query_text_mode_full_compacts_multiline_sql {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        // The pool is constructed before the test telemetry is installed, so the pool's
        // metric instruments bind to the no-op meter; this macro asserts on span-side
        // capture only, in line with `test_query_text_mode_obfuscated_replaces_literals`.
        use sqlx::Executor as _;
        let _ = $dialect;
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw).build();

        let tel = $crate::common::TestTelemetry::install();
        let _row = (&pool)
            .fetch_optional("SELECT\n  1,\n  2,\n  3")
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.query.text"),
            Some(opentelemetry::Value::String("SELECT 1, 2, 3".into())),
            "multi-line SQL must be collapsed to single-space-separated tokens"
        );
    }};
}

/// `QueryTextMode::Obfuscated` emits `db.query.text` with literals replaced by `?` *and*
/// inter-token whitespace collapsed to a single space. The query is dialect-neutral
/// (`SELECT 1, 'alice', 3.14` split across lines), so the only dialect input is the raw
/// pool factory used to build a custom-configured pool.
#[macro_export]
macro_rules! test_query_text_mode_obfuscated_compacts_multiline_sql {
    ($raw_pool_factory:expr, $dialect:expr) => {{
        // Span-side assertion only; metrics are bound to the no-op meter for the same
        // reason as `test_query_text_mode_obfuscated_replaces_literals`.
        use sqlx::Executor as _;
        let _ = $dialect;
        let raw = $raw_pool_factory;
        let pool = sqlx_otel::PoolBuilder::from(raw)
            .with_query_text_mode(sqlx_otel::QueryTextMode::Obfuscated)
            .build();

        let tel = $crate::common::TestTelemetry::install();
        let _row = (&pool)
            .fetch_optional("SELECT 1,\n  'alice',\n  3.14")
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.query.text"),
            Some(opentelemetry::Value::String("SELECT ?, ?, ?".into())),
            "Obfuscated mode must redact literals and collapse multi-line whitespace"
        );
    }};
}

/// `fetch_optional` against an empty table returns `None` and records `returned_rows = 0`.
/// Uses `fresh_table!` to set up a guaranteed-empty table.
#[macro_export]
macro_rules! test_fetch_optional_records_zero_rows {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Executor as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;
        $crate::fresh_table!(
            &pool,
            "empty_table",
            &format!("id {}", $dialect.id_pk_column)
        );
        tel.reset();

        let result = (&pool)
            .fetch_optional("SELECT id FROM empty_table")
            .await
            .unwrap();
        assert!(result.is_none());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_common_span_attributes(&spans[0], $dialect.system);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );

        $crate::common::assert_metric_for_system(&tel, $dialect.system);
    }};
}

/// `sqlx::query(...).bind(...).bind(...).with_annotations(...).fetch_one(&pool)` with a
/// dialect-specific SELECT that adds two bound `i32` arguments and returns an `i64`.
#[macro_export]
macro_rules! test_query_bind_first_then_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let row = sqlx::query($dialect.bind_two_sum_sql)
            .bind(2_i32)
            .bind(3_i32)
            .with_annotations($crate::common::test_annotations())
            .fetch_one(&pool)
            .await
            .unwrap();
        let sum: i64 = row.try_get("sum").unwrap();
        assert_eq!(sum, 5);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Same as `test_query_bind_first_then_annotations_via_pool` but with `with_annotations`
/// applied before the binds.
#[macro_export]
macro_rules! test_query_annotations_first_then_bind_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx::Row as _;
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let row = sqlx::query($dialect.bind_two_sum_sql)
            .with_annotations($crate::common::test_annotations())
            .bind(10_i32)
            .bind(20_i32)
            .fetch_one(&pool)
            .await
            .unwrap();
        let sum: i64 = row.try_get("sum").unwrap();
        assert_eq!(sum, 30);

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Annotated `execute` against the wrapped pool. Uses `SELECT 1` so the test is
/// portable; the executor records `affected_rows` regardless of statement kind. The
/// macro runs the call twice – once inline (documents the simplest usage) and once
/// inside `tokio::spawn` (compile-time proof that the returned future is `Send`, the
/// contract that broke under v0.2.0's `async fn` shape and was restored by converting
/// the forwarder to `fn -> impl Future + Send + 'e`).
#[macro_export]
macro_rules! test_query_execute_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        sqlx::query("SELECT 1")
            .with_annotations($crate::common::test_annotations())
            .execute(&pool)
            .await
            .unwrap();

        let pool_clone = pool.clone();
        tokio::spawn(async move {
            sqlx::query("SELECT 1")
                .with_annotations($crate::common::test_annotations())
                .execute(&pool_clone)
                .await
                .unwrap();
        })
        .await
        .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 2);
        for span in &spans {
            $crate::common::assert_annotated_span(span, &$dialect);
            assert!($crate::common::attr(span, "db.response.affected_rows").is_some());
        }

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Annotated `execute` against `&mut PoolConnection<DB>`. Runs inline and inside
/// `tokio::spawn` to exercise the `Send`-required path over the `&mut PoolConnection`
/// borrow.
#[macro_export]
macro_rules! test_query_execute_with_annotations_via_connection {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut conn = pool.acquire().await.unwrap();
        sqlx::query("SELECT 1")
            .with_annotations($crate::common::test_annotations())
            .execute(&mut conn)
            .await
            .unwrap();
        drop(conn);

        let pool_clone = pool.clone();
        tokio::spawn(async move {
            let mut conn = pool_clone.acquire().await.unwrap();
            sqlx::query("SELECT 1")
                .with_annotations($crate::common::test_annotations())
                .execute(&mut conn)
                .await
                .unwrap();
        })
        .await
        .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 2);
        for span in &spans {
            $crate::common::assert_annotated_span(span, &$dialect);
        }

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Annotated `execute` against `&mut Transaction<'_, DB>`. Runs inline and inside
/// `tokio::spawn` to exercise the `Send`-required path over the `&mut Transaction`
/// borrow.
#[macro_export]
macro_rules! test_query_execute_with_annotations_via_transaction {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let mut tx = pool.begin().await.unwrap();
        sqlx::query("SELECT 1")
            .with_annotations($crate::common::test_annotations())
            .execute(&mut tx)
            .await
            .unwrap();
        tx.commit().await.unwrap();

        let pool_clone = pool.clone();
        tokio::spawn(async move {
            let mut tx = pool_clone.begin().await.unwrap();
            sqlx::query("SELECT 1")
                .with_annotations($crate::common::test_annotations())
                .execute(&mut tx)
                .await
                .unwrap();
            tx.commit().await.unwrap();
        })
        .await
        .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 2);
        for span in &spans {
            $crate::common::assert_annotated_span(span, &$dialect);
        }

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// `with_operation` shorthand attaching the same annotations as the manual
/// `with_annotations(test_annotations())` call.
#[macro_export]
macro_rules! test_query_with_operation_shorthand_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        sqlx::query("SELECT 1")
            .with_operation("SELECT", "users")
            .execute(&pool)
            .await
            .unwrap();

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Annotated `fetch_optional` returning `None`. Uses `SELECT 1 WHERE 1 = 0` to express
/// the empty-row case without dialect-specific table setup.
#[macro_export]
macro_rules! test_query_fetch_optional_with_annotations_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        use sqlx_otel::QueryAnnotateExt as _;
        let tel = $crate::common::TestTelemetry::install();
        let pool = $pool_factory;

        let row = sqlx::query("SELECT 1 WHERE 1 = 0")
            .with_annotations($crate::common::test_annotations())
            .fetch_optional(&pool)
            .await
            .unwrap();
        assert!(row.is_none());

        let spans = tel.spans();
        assert_eq!(spans.len(), 1);
        $crate::common::assert_annotated_span(&spans[0], &$dialect);
        assert_eq!(
            $crate::common::attr(&spans[0], "db.response.returned_rows"),
            Some(opentelemetry::Value::I64(0))
        );

        $crate::common::assert_annotated_metric(&tel, &$dialect);
    }};
}

/// Parity assertion: executor-side and query-side annotation surfaces must produce
/// byte-identical span attributes (per [`crate::query_ext`]'s "Choosing between
/// executor-side and query-side" doc, the two surfaces are documented as semantically
/// equivalent). Both code paths funnel through the same `Annotated<'_, Pool<DB>>`
/// `Executor` impl; if a future refactor diverges them, this test fails. Exercises all
/// three builder families (`query`, `query_as`, `query_scalar`) so a regression in any
/// one of them is caught.
#[macro_export]
macro_rules! test_executor_side_query_side_parity_via_pool {
    ($pool_factory:expr, $dialect:expr) => {{
        // This macro asserts span-side parity between the executor and query annotation
        // surfaces. It does not assert on metrics because the pool is constructed before
        // the test telemetry is installed, so the pool's metric instruments bind to the
        // no-op meter. Per-method metric coverage lives in every other `test_<method>_*`
        // macro.
        use sqlx_otel::QueryAnnotateExt as _;
        let _ = $dialect;
        let pool = $pool_factory;
        let tel = $crate::common::TestTelemetry::install();

        // --- query (execute) -------------------------------------------------
        let _ = sqlx::query("SELECT 1")
            .execute(pool.with_operation("SELECT", "users"))
            .await
            .unwrap();
        let exec_spans = tel.spans();
        tel.reset();

        let _ = sqlx::query("SELECT 1")
            .with_operation("SELECT", "users")
            .execute(&pool)
            .await
            .unwrap();
        let query_spans = tel.spans();
        tel.reset();

        $crate::common::assert_span_parity("query::execute", &exec_spans, &query_spans);

        // --- query_as (fetch_optional) ---------------------------------------
        let _: Option<(i32,)> = sqlx::query_as("SELECT 1")
            .fetch_optional(pool.with_operation("SELECT", "users"))
            .await
            .unwrap();
        let exec_spans = tel.spans();
        tel.reset();

        let _: Option<(i32,)> = sqlx::query_as("SELECT 1")
            .with_operation("SELECT", "users")
            .fetch_optional(&pool)
            .await
            .unwrap();
        let query_spans = tel.spans();
        tel.reset();

        $crate::common::assert_span_parity("query_as::fetch_optional", &exec_spans, &query_spans);

        // --- query_scalar (fetch_one) ----------------------------------------
        let _: i32 = sqlx::query_scalar("SELECT 1")
            .fetch_one(pool.with_operation("SELECT", "users"))
            .await
            .unwrap();
        let exec_spans = tel.spans();
        tel.reset();

        let _: i32 = sqlx::query_scalar("SELECT 1")
            .with_operation("SELECT", "users")
            .fetch_one(&pool)
            .await
            .unwrap();
        let query_spans = tel.spans();

        $crate::common::assert_span_parity("query_scalar::fetch_one", &exec_spans, &query_spans);
    }};
}