refactor(drive): unify AVG no-prove dispatch into a single count+sum walk

book/src/drive/average-index-examples.md

@@ -1702,3 +1702,9 @@ The split closely parallels the count and sum chapters — point lookups for Q1
 The chapter is grounded in the [`document_average_worst_case`](https://github.com/dashpay/platform/blob/v3.1-dev/packages/rs-drive/benches/document_average_worst_case.rs) bench's measured numbers — Q1–Q7 verify cleanly end-to-end against the shared root hash `8b15f732…ffc7`.
 
 A natural expansion follow-up (out of scope here): a worked example of "exact-precision" averages — for callers that need fractional averages (e.g. `avg = 50.7142857…` rather than `50.99`), the protocol-level approach is to return `(count, sum)` and let the client compute in its preferred numeric format (the chapter notes this in [Numerical Considerations](#numerical-considerations) above; a future expansion could walk through the fixed-point vs. floating-point trade-offs).
+
+### No-proof path: single-walk joint dispatch
+
+Both the prove path (above) and the no-proof path now read `(count, sum)` from each visited count-sum-bearing element in a **single** grovedb walk. The no-proof path used to compose parallel count + sum sub-requests and zip the responses; that worked under a shared read transaction (atomic) but did twice the grovedb work strictly necessary. As of [#3687](https://github.com/dashpay/platform/issues/3687) the joint dispatcher at `crate::query::drive_document_count_and_sum_query` walks PCPS / CountSumTree elements once per AVG no-prove query and folds `(count, sum)` in Rust — halving the per-query grovedb work.
+
+One forward-looking optimization remains open in grovedb: a no-prove engine-side `query_aggregate_count_and_sum` accumulator (the proof-side analog `AggregateCountAndSumOnRange` already exists and is what Q5 / Q7 above use). Today's no-prove flat-summed branch walks all matched PCPS elements and folds in Rust; an engine-side accumulator could collapse that fold the way `query_aggregate_sum` collapses the sum-side equivalent. Not a blocker — the Rust-side fold is correctness-equivalent and the per-element count-sum decode is cheap — but it would close the residual perf gap between the no-prove and prove paths.

packages/rs-drive/src/query/drive_document_average_query/mod.rs

@@ -15,15 +15,16 @@
 //! [`book/src/drive/average-index-examples.md`](../../../../../book/src/drive/average-index-examples.md)
 //! for the design and the grades-contract worked example.
 //!
-//! Wired end-to-end: the dispatcher composes the count + sum
-//! executors on the no-proof path under a shared read-transaction
-//! (see [`drive_dispatcher`] module docstring for the atomicity
-//! contract), and the prove path dispatches directly to the PCPS /
-//! primary-key proof executors. A planned follow-up tracked at
-//! [dashpay/platform#3687](https://github.com/dashpay/platform/issues/3687)
-//! will collapse the no-proof path's two-request composition into a
-//! single unified executor that reads both metrics from each visited
-//! PCPS element in one walk.
+//! Wired end-to-end: the dispatcher routes prove-true requests to the
+//! PCPS / primary-key proof executors, and prove-false requests to the
+//! joint single-walk count-and-sum dispatcher at
+//! [`crate::query::drive_document_count_and_sum_query`]. Both paths
+//! read `(count, sum)` from each visited count-sum-bearing element in
+//! a single grovedb walk — see the
+//! [`drive_dispatcher`](drive_dispatcher) module docstring for the
+//! routing details and
+//! [`crate::query::drive_document_count_and_sum_query`] for the
+//! no-prove perf / atomicity contract.
 
 #[cfg(feature = "server")]
 pub mod drive_dispatcher;

packages/rs-drive/src/query/drive_document_count_and_sum_query/executors/mod.rs

@@ -0,0 +1,21 @@
+//! Per-`DocumentSumMode` joint count-and-sum no-prove executors.
+//!
+//! Mirrors [`crate::query::drive_document_sum_query::executors`]
+//! one-to-one for the no-prove subset of modes — the prove modes
+//! (`RangeProof`, `RangeDistinctProof`, `PointLookupProof`,
+//! `RangeAggregateCarrierProof`) are unreachable here because the AVG
+//! dispatcher routes prove-true requests through
+//! [`Drive::execute_document_average_prove`] before this module is
+//! consulted. The three no-prove modes are:
+//!
+//! - [`total`] — `DocumentSumMode::Total` (empty-where +
+//!   `documents_summable + documents_countable` fast path AND
+//!   Equal/In-fully-covered point lookup).
+//! - [`per_in_value`] — `DocumentSumMode::PerInValue` (Equal/In on a
+//!   non-range covered index, emitting one entry per In branch).
+//! - [`range_no_proof`] — `DocumentSumMode::RangeNoProof` (range and
+//!   distinct shapes over a PCPS-eligible index).
+
+pub mod per_in_value;
+pub mod range_no_proof;
+pub mod total;

packages/rs-drive/src/query/drive_document_count_and_sum_query/executors/per_in_value.rs

@@ -0,0 +1,211 @@
+//! Joint count-and-sum `PerInValue` executor for
+//! [`DocumentSumMode::PerInValue`] dispatch on the AVG no-prove path.
+//!
+//! Mirrors [`crate::query::drive_document_sum_query::executors::per_in_value`]
+//! with the same two substitutions documented in the sibling
+//! [`super::total`] module — element decoding via
+//! `count_sum_value_or_default()` and picker-filtering on
+//! `idx.countable.is_countable()`.
+//!
+//! ## Atomicity
+//!
+//! The per-In fan-out issues one grovedb read per In branch. When the
+//! caller didn't provide a transaction, we open a short-lived shared
+//! read transaction internally and reuse it across every per-branch
+//! read so the joint executor sees a single grovedb snapshot. This is
+//! the same atomicity contract the pre-#3687 AVG dispatcher
+//! implemented at the dispatcher layer, moved here because the
+//! dispatcher itself now does only a single executor call per AVG
+//! request — the multi-read concern is purely an internal property of
+//! this executor (and its `range_no_proof` sibling, which has the same
+//! per-In fan-out shape for the compound-flat-summed branch).
+
+use super::super::super::drive_document_average_query::{AverageEntry, DocumentAverageResponse};
+use super::super::super::drive_document_sum_query::index_picker::find_summable_index_for_where_clauses;
+use super::super::super::drive_document_sum_query::{DriveDocumentSumQuery, RangeSumOptions};
+use crate::drive::Drive;
+use crate::error::query::QuerySyntaxError;
+use crate::error::Error;
+use crate::query::{WhereClause, WhereOperator};
+use dpp::data_contract::document_type::DocumentTypeRef;
+use dpp::version::PlatformVersion;
+use grovedb::query_result_type::{QueryResultElement, QueryResultType};
+use grovedb::TransactionArg;
+
+impl Drive {
+    /// Cartesian-forks the single `In` clause into one Equal-per-value
+    /// sub-query against a `summable + countable` index; reads
+    /// `(count, sum)` from each per-branch point-lookup walk in one
+    /// grovedb read each, then emits per-In-value
+    /// [`AverageEntry { in_key: None, key, count, sum }`] entries.
+    ///
+    /// Returns [`DocumentAverageResponse::Entries`].
+    #[allow(clippy::too_many_arguments)]
+    pub fn execute_document_count_and_sum_per_in_value_no_proof(
+        &self,
+        contract_id: [u8; 32],
+        document_type: DocumentTypeRef,
+        document_type_name: String,
+        where_clauses: Vec<WhereClause>,
+        sum_property: String,
+        options: RangeSumOptions,
+        limit: Option<u32>,
+        transaction: TransactionArg,
+        platform_version: &PlatformVersion,
+    ) -> Result<DocumentAverageResponse, Error> {
+        // Enforce exactly one `In` clause. The sum-side analog
+        // documents the silent-drop bug this guards against; same
+        // failure mode would apply here.
+        let in_clauses: Vec<&WhereClause> = where_clauses
+            .iter()
+            .filter(|wc| wc.operator == WhereOperator::In)
+            .collect();
+        if in_clauses.len() != 1 {
+            return Err(Error::Query(
+                QuerySyntaxError::InvalidWhereClauseComponents(
+                    "execute_document_count_and_sum_per_in_value_no_proof requires \
+                     exactly one `in` clause",
+                ),
+            ));
+        }
+        let in_clause = in_clauses[0].clone();
+        let in_values = in_clause.in_values().into_data_with_error()??;
+
+        let other_clauses: Vec<WhereClause> = where_clauses
+            .iter()
+            .filter(|wc| wc.operator != WhereOperator::In)
+            .cloned()
+            .collect();
+
+        // Open a short-lived shared read transaction if the caller
+        // didn't provide one. Multiple per-In branches read grovedb
+        // separately; without a shared snapshot a concurrent block
+        // commit could slip between branches and produce a
+        // `(count, sum)` pair from inconsistent state. Read-only; the
+        // local transaction is dropped without commit at the end.
+        let local_tx;
+        let effective_transaction: TransactionArg = if transaction.is_some() {
+            transaction
+        } else {
+            local_tx = self.grove.start_transaction();
+            Some(&local_tx)
+        };
+
+        use dpp::data_contract::document_type::accessors::DocumentTypeV0Getters;
+        use dpp::data_contract::document_type::methods::DocumentTypeV0Methods;
+        // BTreeMap deduplicates by canonical key bytes and orders
+        // ascending — matches sum-side analog. Storing the full
+        // (count, sum) pair per key rather than separate maps means
+        // we never need to zip post-walk; each grovedb read writes
+        // exactly one entry into the map.
+        let mut merged: std::collections::BTreeMap<Vec<u8>, (u64, i64)> =
+            std::collections::BTreeMap::new();
+        for value in in_values.iter() {
+            let key_bytes = document_type.serialize_value_for_key(
+                in_clause.field.as_str(),
+                value,
+                platform_version,
+            )?;
+            if merged.contains_key(&key_bytes) {
+                continue;
+            }
+
+            let mut clauses_for_value = other_clauses.clone();
+            clauses_for_value.push(WhereClause {
+                field: in_clause.field.clone(),
+                operator: WhereOperator::Equal,
+                value: value.clone(),
+            });
+
+            let index = find_summable_index_for_where_clauses(
+                document_type.indexes(),
+                &clauses_for_value,
+                &sum_property,
+            )
+            .filter(|idx| idx.countable.is_countable())
+            .ok_or_else(|| {
+                Error::Query(QuerySyntaxError::WhereClauseOnNonIndexedProperty(
+                    "average query requires an index that declares BOTH \
+                     `summable: \"<prop>\"` AND a countable terminator \
+                     (`countable: \"countable\"` or `\"countableAllowingOffset\"`) \
+                     matching the where clause fields"
+                        .to_string(),
+                ))
+            })?;
+
+            let sum_query = DriveDocumentSumQuery {
+                document_type,
+                contract_id,
+                document_type_name: document_type_name.clone(),
+                index,
+                where_clauses: clauses_for_value,
+                sum_property: sum_property.clone(),
+            };
+
+            let drive_version = &platform_version.drive;
+            let path_query = sum_query.point_lookup_sum_path_query(platform_version)?;
+            let mut drive_operations = vec![];
+            let (results, _) = self.grove_get_path_query(
+                &path_query,
+                effective_transaction,
+                QueryResultType::QueryElementResultType,
+                &mut drive_operations,
+                drive_version,
+            )?;
+
+            // For the per-In-value branch the inner walk visits at
+            // most one count-sum-bearing element (the Equal arm fully
+            // covered the index). Sum across the emitted elements to
+            // handle the boundary case where the picker accepts an
+            // index whose terminator is `In`-shaped in the same prop
+            // (rare for AVG since the dispatched In was lifted out,
+            // but cheap to do correctly).
+            let mut count_acc: u64 = 0;
+            let mut sum_acc: i64 = 0;
+            for elem in results.elements.iter() {
+                if let QueryResultElement::ElementResultItem(element) = elem {
+                    let (c, s) = element.count_sum_value_or_default();
+                    count_acc = count_acc.checked_add(c).ok_or_else(|| {
+                        Error::Query(QuerySyntaxError::Unsupported(
+                            "per-In-value count-and-sum overflowed u64 on the count axis \
+                             when summing branch elements. Narrow the query or use \
+                             multiple queries and combine client-side."
+                                .to_string(),
+                        ))
+                    })?;
+                    sum_acc = sum_acc.checked_add(s).ok_or_else(|| {
+                        Error::Query(QuerySyntaxError::Unsupported(
+                            "per-In-value count-and-sum overflowed i64 on the sum axis \
+                             when summing branch elements. Narrow the query or use \
+                             multiple queries and combine client-side."
+                                .to_string(),
+                        ))
+                    })?;
+                }
+            }
+            merged.insert(key_bytes, (count_acc, sum_acc));
+        }
+
+        let mut entries: Vec<AverageEntry> = merged
+            .into_iter()
+            .map(|(key, (count, sum))| AverageEntry {
+                in_key: None,
+                key,
+                count: Some(count),
+                sum: Some(sum),
+            })
+            .collect();
+        if !options.left_to_right {
+            entries.reverse();
+        }
+        // Apply caller's `limit` AFTER ordering — same shape as
+        // count's `execute_document_count_per_in_value_no_proof`. The
+        // BTreeMap iteration was already ascending; an explicit
+        // descending order reverses the vec, then the truncate caps
+        // it at `limit` from the front.
+        if let Some(l) = limit {
+            entries.truncate(l as usize);
+        }
+        Ok(DocumentAverageResponse::Entries(entries))
+    }
+}