feat: Record db.client.response.affected_rows histogram

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 
 ### Added
 
+- `db.client.response.affected_rows` histogram – a custom metric (no OTel semconv equivalent) mirroring the existing `db.response.affected_rows` span attribute. Recorded on `execute()` calls; carries the same connection / annotation / error attribute set as the duration histogram so dashboards can slice mutation throughput by the same dimensions ([#33](https://github.com/chmodas/sqlx-otel/pull/33)).
 - `PoolBuilder::with_network_protocol_name` and `with_network_transport` builder methods, plus a per-backend `Database::DEFAULT_NETWORK_PROTOCOL_NAME` constant (Postgres → `"postgresql"`, MySQL → `"mysql"`, SQLite → `None`). `network.protocol.name`, `network.transport`, and `db.client.connection.pool.name` now surface on every span and per-operation metric data point so dashboards can slice query latency by the same dimensions OTel's database-spans semconv recommends ([#32](https://github.com/chmodas/sqlx-otel/pull/32)).
 
 ### Changed

README.md

@@ -168,13 +168,16 @@ On error, the span status is set to `Error` and an `exception` event is added wi
 
 ### Operation metrics
 
-| Instrument                         | Type      | Unit | Description                           |
-|------------------------------------|-----------|------|---------------------------------------|
-| `db.client.operation.duration`     | Histogram | `s`  | Duration of each database operation   |
-| `db.client.response.returned_rows` | Histogram |      | Number of rows returned per operation |
+| Instrument                         | Type      | Unit | Description                                   |
+|------------------------------------|-----------|------|-----------------------------------------------|
+| `db.client.operation.duration`     | Histogram | `s`  | Duration of each database operation           |
+| `db.client.response.returned_rows` | Histogram |      | Number of rows returned per `fetch*` call     |
+| `db.client.response.affected_rows` | Histogram |      | Rows affected per `execute` call (custom)     |
 
 These mirror the bounded portion of the span attribute set: connection-level attributes (`db.system.name`, `db.namespace`, `server.address`/`port`, `network.peer.address`/`port`, `network.protocol.name`, `network.transport`, `db.client.connection.pool.name` – wherever set), plus annotation-derived attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) when present, plus error-path attributes (`error.type`, plus `db.response.status_code` for `sqlx::Error::Database`) on the error path. `db.query.text` is deliberately excluded for cardinality; `db.query.summary` is caller-controlled and inherits its cardinality cost from the span side.
 
+`db.client.response.affected_rows` is not part of the OpenTelemetry semantic conventions – we ship it for the same reason as the matching span attribute: backends report a useful database-confirmed count that's worth slicing alongside duration. It is recorded only on `execute()` calls (where `QueryResult::rows_affected()` is meaningful) and is not recorded for `execute_many` (deprecated upstream).
+
 ### Connection pool metrics
 
 | Instrument                                | Type            | Unit | Description                                           |

src/executor.rs

@@ -94,7 +94,7 @@ fn start_span(name: &str, span_attrs: Vec<KeyValue>) -> (OtelContext, Instant) {
 /// attributes plus the four annotation-derived attributes when present, plus error-path attributes
 /// (`error.type`, `db.response.status_code`) appended later by `record_error`. The unbounded
 /// `db.query.text` attribute is deliberately excluded; `db.query.summary` is caller-controlled and
-/// can be unbounded — that cardinality cost is inherited from the span side.
+/// can be unbounded – that cardinality cost is inherited from the span side.
 fn begin_query_span(
     attrs: &ConnectionAttributes,
     sql: Option<&str>,
@@ -187,16 +187,19 @@ fn record_affected_rows(cx: &OtelContext, rows: u64) {
     ));
 }
 
-/// End the span and record metrics.
+/// End the span and record metrics. `returned_rows` is `Some` for `fetch*` paths,
+/// `affected_rows` is `Some` for `execute` paths; both are `None` for paths that report
+/// neither (e.g. `prepare` / `describe` / `execute_many`'s streaming aggregate).
 fn finish(
     cx: &OtelContext,
     start: Instant,
-    rows: Option<u64>,
+    returned_rows: Option<u64>,
+    affected_rows: Option<u64>,
     metrics: &Metrics,
     attrs: &[KeyValue],
 ) {
     cx.span().end();
-    metrics.record(start.elapsed(), rows, attrs);
+    metrics.record(start.elapsed(), returned_rows, affected_rows, attrs);
 }
 
 /// Await a future, record any error on the span, then finish. Used by `execute`, `prepare`,
@@ -212,7 +215,7 @@ async fn execute_instrumented<T>(
     if let Err(err) = &result {
         record_error(&cx, err, &mut metric_attrs);
     }
-    finish(&cx, start, None, &metrics, &metric_attrs);
+    finish(&cx, start, None, None, &metrics, &metric_attrs);
     result
 }
 
@@ -297,6 +300,7 @@ impl<S, C> InstrumentedStream<S, C> {
                 &self.cx,
                 self.start,
                 Some(self.rows),
+                None,
                 &self.metrics,
                 &self.metric_attrs,
             );
@@ -395,15 +399,18 @@ macro_rules! impl_executor {
                 let fut = ($inner).execute(query);
                 Box::pin(async move {
                     let result = fut.await;
-                    match &result {
+                    let affected = match &result {
                         Ok(qr) => {
-                            record_affected_rows(&cx, DB::rows_affected(qr));
+                            let n = DB::rows_affected(qr);
+                            record_affected_rows(&cx, n);
+                            Some(n)
                         }
                         Err(err) => {
                             record_error(&cx, err, &mut metric_attrs);
+                            None
                         }
-                    }
-                    finish(&cx, start, None, &state.metrics, &metric_attrs);
+                    };
+                    finish(&cx, start, None, affected, &state.metrics, &metric_attrs);
                     result
                 })
             }
@@ -512,11 +519,11 @@ macro_rules! impl_executor {
                         Ok(rows) => {
                             let count = rows.len() as u64;
                             record_rows(&cx, count);
-                            finish(&cx, start, Some(count), &state.metrics, &metric_attrs);
+                            finish(&cx, start, Some(count), None, &state.metrics, &metric_attrs);
                         }
                         Err(err) => {
                             record_error(&cx, err, &mut metric_attrs);
-                            finish(&cx, start, None, &state.metrics, &metric_attrs);
+                            finish(&cx, start, None, None, &state.metrics, &metric_attrs);
                         }
                     }
                     result
@@ -545,11 +552,11 @@ macro_rules! impl_executor {
                     match &result {
                         Ok(_) => {
                             record_rows(&cx, 1);
-                            finish(&cx, start, Some(1), &state.metrics, &metric_attrs);
+                            finish(&cx, start, Some(1), None, &state.metrics, &metric_attrs);
                         }
                         Err(err) => {
                             record_error(&cx, err, &mut metric_attrs);
-                            finish(&cx, start, None, &state.metrics, &metric_attrs);
+                            finish(&cx, start, None, None, &state.metrics, &metric_attrs);
                         }
                     }
                     result
@@ -579,11 +586,11 @@ macro_rules! impl_executor {
                         Ok(maybe_row) => {
                             let count = u64::from(maybe_row.is_some());
                             record_rows(&cx, count);
-                            finish(&cx, start, Some(count), &state.metrics, &metric_attrs);
+                            finish(&cx, start, Some(count), None, &state.metrics, &metric_attrs);
                         }
                         Err(err) => {
                             record_error(&cx, err, &mut metric_attrs);
-                            finish(&cx, start, None, &state.metrics, &metric_attrs);
+                            finish(&cx, start, None, None, &state.metrics, &metric_attrs);
                         }
                     }
                     result
@@ -1177,7 +1184,7 @@ mod tests {
         /// the appended key set is exactly `{"db.operation.name" iff op.is_some(),
         /// "db.collection.name" iff coll.is_some(), "db.query.summary" iff
         /// query_summary.is_some(), "db.stored_procedure.name" iff
-        /// stored_procedure.is_some()}` — and nothing else, in particular none of the
+        /// stored_procedure.is_some()}` – and nothing else, in particular none of the
         /// connection or query-text keys leak through.
         #[test]
         fn append_annotation_attrs_membership_invariant(ann in any_annotations()) {

src/metrics.rs

@@ -13,6 +13,11 @@ use opentelemetry_semantic_conventions::metric;
 pub(crate) struct Metrics {
     duration: Histogram<f64>,
     returned_rows: Histogram<f64>,
+    /// Custom histogram (no `OTel` semconv equivalent) recording the database-confirmed
+    /// `rows_affected()` count for `execute()` operations. Mirrors the existing
+    /// `db.response.affected_rows` span attribute so dashboards can slice mutation
+    /// throughput by the same dimensions.
+    affected_rows: Histogram<f64>,
 }
 
 impl Metrics {
@@ -28,18 +33,38 @@ impl Metrics {
             .f64_histogram(metric::DB_CLIENT_RESPONSE_RETURNED_ROWS)
             .with_description("Number of rows returned by database operations.")
             .build();
+        let affected_rows = meter
+            .f64_histogram("db.client.response.affected_rows")
+            .with_description("Number of rows affected by database operations.")
+            .build();
         Self {
             duration,
             returned_rows,
+            affected_rows,
         }
     }
 
-    /// Record a completed operation's duration and, optionally, the number of rows returned.
-    pub fn record(&self, elapsed: Duration, rows: Option<u64>, attributes: &[KeyValue]) {
+    /// Record a completed operation: always the duration histogram; `returned_rows` and
+    /// `affected_rows` histograms when their respective counts are `Some`. The two row-
+    /// count parameters are mutually exclusive in practice (a `fetch*` operation sets
+    /// `returned_rows`; an `execute` operation sets `affected_rows`), but the signature
+    /// allows both for forward compatibility with backends that report both for a single
+    /// operation.
+    pub fn record(
+        &self,
+        elapsed: Duration,
+        returned_rows: Option<u64>,
+        affected_rows: Option<u64>,
+        attributes: &[KeyValue],
+    ) {
         self.duration.record(elapsed.as_secs_f64(), attributes);
-        if let Some(count) = rows {
+        if let Some(count) = returned_rows {
             #[allow(clippy::cast_precision_loss)]
             self.returned_rows.record(count as f64, attributes);
         }
+        if let Some(count) = affected_rows {
+            #[allow(clippy::cast_precision_loss)]
+            self.affected_rows.record(count as f64, attributes);
+        }
     }
 }

tests/common.rs

@@ -102,6 +102,65 @@ pub fn metric_attr(
         .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.
 ///
@@ -135,7 +194,7 @@ pub fn find_duration_data_point(
 ///
 /// 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.
+/// 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)],
@@ -2277,7 +2336,7 @@ macro_rules! test_operation_duration_metric_carries_full_annotations {
 /// 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`
+/// `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.
@@ -3575,6 +3634,7 @@ macro_rules! test_execute_records_affected_rows {
             "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) ---
@@ -3586,6 +3646,7 @@ macro_rules! test_execute_records_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) ---
@@ -3600,6 +3661,7 @@ macro_rules! test_execute_records_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 ---
@@ -3614,6 +3676,7 @@ macro_rules! test_execute_records_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 ---
@@ -3629,6 +3692,7 @@ macro_rules! test_execute_records_affected_rows {
             "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);
     }};
 }