diff --git a/components/zcash_protocol/CHANGELOG.md b/components/zcash_protocol/CHANGELOG.md
index 963652662..3291d5f9d 100644
--- a/components/zcash_protocol/CHANGELOG.md
+++ b/components/zcash_protocol/CHANGELOG.md
@@ -10,6 +10,8 @@ and this library adheres to Rust's notion of
 ### Added
 - `zcash_protocol::value::QuotRem`
 - `zcash_protocol::value::Zatoshis::div_with_remainder`
+- `impl Mul<u64> for zcash_protocol::value::Zatoshis`
+- `impl Div<NonZeroU64> for zcash_protocol::value::Zatoshis`
 
 ### Changed
 - MSRV is now 1.77.0.
diff --git a/components/zcash_protocol/src/value.rs b/components/zcash_protocol/src/value.rs
index 94f2b7ba6..aa4d1709b 100644
--- a/components/zcash_protocol/src/value.rs
+++ b/components/zcash_protocol/src/value.rs
@@ -2,7 +2,7 @@ use std::convert::{Infallible, TryFrom};
 use std::error;
 use std::iter::Sum;
 use std::num::NonZeroU64;
-use std::ops::{Add, Mul, Neg, Sub};
+use std::ops::{Add, Div, Mul, Neg, Sub};
 
 use memuse::DynamicUsage;
 
@@ -321,6 +321,7 @@ impl Zatoshis {
     /// Divides this `Zatoshis` value by the given divisor and returns the quotient and remainder.
     pub fn div_with_remainder(&self, divisor: NonZeroU64) -> QuotRem<Zatoshis> {
         let divisor = u64::from(divisor);
+        // `self` is already bounds-checked, so we don't need to re-check it in division
         QuotRem {
             quotient: Zatoshis(self.0 / divisor),
             remainder: Zatoshis(self.0 % divisor),
@@ -394,11 +395,19 @@ impl Sub<Zatoshis> for Option<Zatoshis> {
     }
 }
 
+impl Mul<u64> for Zatoshis {
+    type Output = Option<Self>;
+
+    fn mul(self, rhs: u64) -> Option<Zatoshis> {
+        Zatoshis::from_u64(self.0.checked_mul(rhs)?).ok()
+    }
+}
+
 impl Mul<usize> for Zatoshis {
     type Output = Option<Self>;
 
     fn mul(self, rhs: usize) -> Option<Zatoshis> {
-        Zatoshis::from_u64(self.0.checked_mul(u64::try_from(rhs).ok()?)?).ok()
+        self * u64::try_from(rhs).ok()?
     }
 }
 
@@ -414,6 +423,15 @@ impl<'a> Sum<&'a Zatoshis> for Option<Zatoshis> {
     }
 }
 
+impl Div<NonZeroU64> for Zatoshis {
+    type Output = Zatoshis;
+
+    fn div(self, rhs: NonZeroU64) -> Zatoshis {
+        // `self` is already bounds-checked, so we don't need to re-check it
+        Zatoshis(self.0 / u64::from(rhs))
+    }
+}
+
 /// A type for balance violations in amount addition and subtraction
 /// (overflow and underflow of allowed ranges)
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
diff --git a/zcash_client_backend/CHANGELOG.md b/zcash_client_backend/CHANGELOG.md
index aebf46076..a915fd569 100644
--- a/zcash_client_backend/CHANGELOG.md
+++ b/zcash_client_backend/CHANGELOG.md
@@ -11,8 +11,11 @@ and this library adheres to Rust's notion of
 - `zcash_client_backend::data_api`:
   - `Progress`
   - `WalletSummary::progress`
-  - `WalletMeta`
+  - `PoolMeta`
+  - `AccountMeta`
   - `impl Default for wallet::input_selection::GreedyInputSelector`
+  - `BoundedU8`
+  - `NoteFilter`
 - `zcash_client_backend::fees`
   - `SplitPolicy`
   - `StandardFeeRule` has been moved here from `zcash_primitives::fees`. Relative
@@ -32,7 +35,7 @@ and this library adheres to Rust's notion of
 - MSRV is now 1.77.0.
 - Migrated to `arti-client 0.23`.
 - `zcash_client_backend::data_api`:
-  - `InputSource` has an added method `get_wallet_metadata`
+  - `InputSource` has an added method `get_account_metadata`
   - `error::Error` has additional variant `Error::Change`. This necessitates
     the addition of two type parameters to the `Error` type,
     `ChangeErrT` and `NoteRefT`.
@@ -65,7 +68,7 @@ and this library adheres to Rust's notion of
     changed.
 - `zcash_client_backend::fees`:
   - `ChangeStrategy` has changed. It has two new associated types, `MetaSource`
-    and `WalletMeta`, and its `FeeRule` associated type now has an additional
+    and `AccountMetaT`, and its `FeeRule` associated type now has an additional
     `Clone` bound. In addition, it defines a new `fetch_wallet_meta` method, and
     the arguments to `compute_balance` have changed.
   - `zip317::SingleOutputChangeStrategy` has been made polymorphic in the fee
diff --git a/zcash_client_backend/src/data_api.rs b/zcash_client_backend/src/data_api.rs
index ab827f70a..a2c9a635a 100644
--- a/zcash_client_backend/src/data_api.rs
+++ b/zcash_client_backend/src/data_api.rs
@@ -794,6 +794,35 @@ impl<NoteRef> SpendableNotes<NoteRef> {
     }
 }
 
+/// Metadata about the structure of unspent outputs in a single pool within a wallet account.
+///
+/// This type is often used to represent a filtered view of outputs in the account that were
+/// selected according to the conditions imposed by a [`NoteFilter`].
+#[derive(Debug, Clone)]
+pub struct PoolMeta {
+    note_count: usize,
+    value: NonNegativeAmount,
+}
+
+impl PoolMeta {
+    /// Constructs a new [`PoolMeta`] value from its constituent parts.
+    pub fn new(note_count: usize, value: NonNegativeAmount) -> Self {
+        Self { note_count, value }
+    }
+
+    /// Returns the number of unspent outputs in the account, potentially selected in accordance
+    /// with some [`NoteFilter`].
+    pub fn note_count(&self) -> usize {
+        self.note_count
+    }
+
+    /// Returns the total value of unspent outputs in the account that are accounted for in
+    /// [`Self::note_count`].
+    pub fn value(&self) -> NonNegativeAmount {
+        self.value
+    }
+}
+
 /// Metadata about the structure of the wallet for a particular account.
 ///
 /// At present this just contains counts of unspent outputs in each pool, but it may be extended in
@@ -802,58 +831,185 @@ impl<NoteRef> SpendableNotes<NoteRef> {
 /// Values of this type are intended to be used in selection of change output values. A value of
 /// this type may represent filtered data, and may therefore not count all of the unspent notes in
 /// the wallet.
-pub struct WalletMeta {
-    sapling_note_count: usize,
-    #[cfg(feature = "orchard")]
-    orchard_note_count: usize,
+///
+/// A [`AccountMeta`] value is normally produced by querying the wallet database via passing a
+/// [`NoteFilter`] to [`InputSource::get_account_metadata`].
+#[derive(Debug, Clone)]
+pub struct AccountMeta {
+    sapling: Option<PoolMeta>,
+    orchard: Option<PoolMeta>,
 }
 
-impl WalletMeta {
-    /// Constructs a new [`WalletMeta`] value from its constituent parts.
-    pub fn new(
-        sapling_note_count: usize,
-        #[cfg(feature = "orchard")] orchard_note_count: usize,
-    ) -> Self {
-        Self {
-            sapling_note_count,
-            #[cfg(feature = "orchard")]
-            orchard_note_count,
-        }
+impl AccountMeta {
+    /// Constructs a new [`AccountMeta`] value from its constituent parts.
+    pub fn new(sapling: Option<PoolMeta>, orchard: Option<PoolMeta>) -> Self {
+        Self { sapling, orchard }
+    }
+
+    /// Returns metadata about Sapling notes belonging to the account for which this was generated.
+    ///
+    /// Returns [`None`] if no metadata is available or it was not possible to evaluate the query
+    /// described by a [`NoteFilter`] given the available wallet data.
+    pub fn sapling(&self) -> Option<&PoolMeta> {
+        self.sapling.as_ref()
+    }
+
+    /// Returns metadata about Orchard notes belonging to the account for which this was generated.
+    ///
+    /// Returns [`None`] if no metadata is available or it was not possible to evaluate the query
+    /// described by a [`NoteFilter`] given the available wallet data.
+    pub fn orchard(&self) -> Option<&PoolMeta> {
+        self.orchard.as_ref()
+    }
+
+    fn sapling_note_count(&self) -> Option<usize> {
+        self.sapling.as_ref().map(|m| m.note_count)
+    }
+
+    fn orchard_note_count(&self) -> Option<usize> {
+        self.orchard.as_ref().map(|m| m.note_count)
     }
 
     /// Returns the number of unspent notes in the wallet for the given shielded protocol.
-    pub fn note_count(&self, protocol: ShieldedProtocol) -> usize {
+    pub fn note_count(&self, protocol: ShieldedProtocol) -> Option<usize> {
         match protocol {
-            ShieldedProtocol::Sapling => self.sapling_note_count,
-            #[cfg(feature = "orchard")]
-            ShieldedProtocol::Orchard => self.orchard_note_count,
-            #[cfg(not(feature = "orchard"))]
-            ShieldedProtocol::Orchard => 0,
+            ShieldedProtocol::Sapling => self.sapling_note_count(),
+            ShieldedProtocol::Orchard => self.orchard_note_count(),
+        }
+    }
+
+    /// Returns the total number of unspent shielded notes belonging to the account for which this
+    /// was generated.
+    ///
+    /// Returns [`None`] if no metadata is available or it was not possible to evaluate the query
+    /// described by a [`NoteFilter`] given the available wallet data. If metadata is available
+    /// only for a single pool, the metadata for that pool will be returned.
+    pub fn total_note_count(&self) -> Option<usize> {
+        let s = self.sapling_note_count();
+        let o = self.orchard_note_count();
+        s.zip(o).map(|(s, o)| s + o).or(s).or(o)
+    }
+
+    fn sapling_value(&self) -> Option<NonNegativeAmount> {
+        self.sapling.as_ref().map(|m| m.value)
+    }
+
+    fn orchard_value(&self) -> Option<NonNegativeAmount> {
+        self.orchard.as_ref().map(|m| m.value)
+    }
+
+    /// Returns the total value of shielded notes represented by [`Self::total_note_count`]
+    ///
+    /// Returns [`None`] if no metadata is available or it was not possible to evaluate the query
+    /// described by a [`NoteFilter`] given the available wallet data. If metadata is available
+    /// only for a single pool, the metadata for that pool will be returned.
+    pub fn total_value(&self) -> Option<NonNegativeAmount> {
+        let s = self.sapling_value();
+        let o = self.orchard_value();
+        s.zip(o)
+            .map(|(s, o)| (s + o).expect("Does not overflow Zcash maximum value."))
+            .or(s)
+            .or(o)
+    }
+}
+
+/// A `u8` value in the range 0..=MAX
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub struct BoundedU8<const MAX: u8>(u8);
+
+impl<const MAX: u8> BoundedU8<MAX> {
+    /// Creates a constant `BoundedU8` from a [`u8`] value.
+    ///
+    /// Panics: if the value is outside the range `0..=MAX`.
+    pub const fn new_const(value: u8) -> Self {
+        assert!(value <= MAX);
+        Self(value)
+    }
+
+    /// Creates a `BoundedU8` from a [`u8`] value.
+    ///
+    /// Returns `None` if the provided value is outside the range `0..=MAX`.
+    pub fn new(value: u8) -> Option<Self> {
+        if value <= MAX {
+            Some(Self(value))
+        } else {
+            None
         }
     }
 
-    /// Returns the number of unspent Sapling notes belonging to the account for which this was
-    /// generated.
-    pub fn sapling_note_count(&self) -> usize {
-        self.sapling_note_count
+    /// Returns the wrapped [`u8`] value.
+    pub fn value(&self) -> u8 {
+        self.0
     }
+}
 
-    /// Returns the number of unspent Orchard notes belonging to the account for which this was
-    /// generated.
-    #[cfg(feature = "orchard")]
-    pub fn orchard_note_count(&self) -> usize {
-        self.orchard_note_count
+impl<const MAX: u8> From<BoundedU8<MAX>> for u8 {
+    fn from(value: BoundedU8<MAX>) -> Self {
+        value.0
     }
+}
 
-    /// Returns the total number of unspent shielded notes belonging to the account for which this
-    /// was generated.
-    pub fn total_note_count(&self) -> usize {
-        self.sapling_note_count + self.note_count(ShieldedProtocol::Orchard)
+impl<const MAX: u8> From<BoundedU8<MAX>> for usize {
+    fn from(value: BoundedU8<MAX>) -> Self {
+        usize::from(value.0)
+    }
+}
+
+/// A small query language for filtering notes belonging to an account.
+///
+/// A filter described using this language is applied to notes individually. It is primarily
+/// intended for retrieval of account metadata in service of making determinations for how to
+/// allocate change notes, and is not currently intended for use in broader note selection
+/// contexts.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum NoteFilter {
+    /// Selects notes having value greater than or equal to the provided value.
+    ExceedsMinValue(NonNegativeAmount),
+    /// Selects notes having value greater than or equal to approximately the n'th percentile of
+    /// previously sent notes in the account, irrespective of pool. The wrapped value must be in
+    /// the range `1..=99`. The value `n` is respected in a best-effort fashion; results are likely
+    /// to be inaccurate if the account has not yet completed scanning or if insufficient send data
+    /// is available to establish a distribution.
+    // TODO: it might be worthwhile to add an optional parameter here that can be used to ignore
+    // low-valued (test/memo-only) sends when constructing the distribution to be drawn from.
+    ExceedsPriorSendPercentile(BoundedU8<99>),
+    /// Selects notes having value greater than or equal to the specified percentage of the account
+    /// balance across all shielded pools. The wrapped value must be in the range `1..=99`
+    ExceedsBalancePercentage(BoundedU8<99>),
+    /// A note will be selected if it satisfies both of the specified conditions.
+    ///
+    /// If it is not possible to evaluate one of the conditions (for example,
+    /// [`NoteFilter::ExceedsPriorSendPercentile`] cannot be evaluated if no sends have been
+    /// performed) then that condition will be ignored. If neither condition can be evaluated,
+    /// then the entire condition cannot be evaluated.
+    Combine(Box<NoteFilter>, Box<NoteFilter>),
+    /// A note will be selected if it satisfies the first condition; if it is not possible to
+    /// evaluate that condition (for example, [`NoteFilter::ExceedsPriorSendPercentile`] cannot
+    /// be evaluated if no sends have been performed) then the second condition will be used for
+    /// evaluation.
+    Attempt {
+        condition: Box<NoteFilter>,
+        fallback: Box<NoteFilter>,
+    },
+}
+
+impl NoteFilter {
+    /// Constructs a [`NoteFilter::Combine`] query node.
+    pub fn combine(l: NoteFilter, r: NoteFilter) -> Self {
+        Self::Combine(Box::new(l), Box::new(r))
+    }
+
+    /// Constructs a [`NoteFilter::Attempt`] query node.
+    pub fn attempt(condition: NoteFilter, fallback: NoteFilter) -> Self {
+        Self::Attempt {
+            condition: Box::new(condition),
+            fallback: Box::new(fallback),
+        }
     }
 }
 
 /// A trait representing the capability to query a data store for unspent transaction outputs
-/// belonging to a wallet.
+/// belonging to a account.
 #[cfg_attr(feature = "test-dependencies", delegatable_trait)]
 pub trait InputSource {
     /// The type of errors produced by a wallet backend.
@@ -900,14 +1056,17 @@ pub trait InputSource {
     ///
     /// The returned metadata value must exclude:
     /// - spent notes;
-    /// - unspent notes having value less than the specified minimum value;
+    /// - unspent notes excluded by the provided selector;
     /// - unspent notes identified in the given `exclude` list.
-    fn get_wallet_metadata(
+    ///
+    /// Implementations of this method may limit the complexity of supported queries. Such
+    /// limitations should be clearly documented for the implementing type.
+    fn get_account_metadata(
         &self,
         account: Self::AccountId,
-        min_value: NonNegativeAmount,
+        selector: &NoteFilter,
         exclude: &[Self::NoteRef],
-    ) -> Result<WalletMeta, Self::Error>;
+    ) -> Result<AccountMeta, Self::Error>;
 
     /// Fetches the transparent output corresponding to the provided `outpoint`.
     ///
diff --git a/zcash_client_backend/src/data_api/testing.rs b/zcash_client_backend/src/data_api/testing.rs
index 0c117c7c8..0041c24bc 100644
--- a/zcash_client_backend/src/data_api/testing.rs
+++ b/zcash_client_backend/src/data_api/testing.rs
@@ -23,6 +23,7 @@ use secrecy::{ExposeSecret, Secret, SecretVec};
 use shardtree::{error::ShardTreeError, store::memory::MemoryShardStore, ShardTree};
 use subtle::ConditionallySelectable;
 
+use zcash_address::ZcashAddress;
 use zcash_keys::address::Address;
 use zcash_note_encryption::Domain;
 use zcash_primitives::{
@@ -43,6 +44,7 @@ use zcash_protocol::{
     value::{ZatBalance, Zatoshis},
 };
 use zip32::{fingerprint::SeedFingerprint, DiversifierIndex};
+use zip321::Payment;
 
 use crate::{
     address::UnifiedAddress,
@@ -59,7 +61,6 @@ use crate::{
     ShieldedProtocol,
 };
 
-use super::error::Error;
 use super::{
     chain::{scan_cached_blocks, BlockSource, ChainState, CommitmentTreeRoot, ScanSummary},
     scanning::ScanRange,
@@ -68,12 +69,13 @@ use super::{
         input_selection::{GreedyInputSelector, InputSelector},
         propose_standard_transfer_to_address, propose_transfer,
     },
-    Account, AccountBalance, AccountBirthday, AccountPurpose, AccountSource, BlockMetadata,
-    DecryptedTransaction, InputSource, NullifierQuery, ScannedBlock, SeedRelevance,
+    Account, AccountBalance, AccountBirthday, AccountMeta, AccountPurpose, AccountSource,
+    BlockMetadata, DecryptedTransaction, InputSource, NullifierQuery, ScannedBlock, SeedRelevance,
     SentTransaction, SpendableNotes, TransactionDataRequest, TransactionStatus,
-    WalletCommitmentTrees, WalletMeta, WalletRead, WalletSummary, WalletTest, WalletWrite,
+    WalletCommitmentTrees, WalletRead, WalletSummary, WalletTest, WalletWrite,
     SAPLING_SHARD_HEIGHT,
 };
+use super::{error::Error, NoteFilter};
 
 #[cfg(feature = "transparent-inputs")]
 use {
@@ -861,6 +863,49 @@ where
         + WalletCommitmentTrees,
     <DbT as WalletRead>::AccountId: ConditionallySelectable + Default + Send + 'static,
 {
+    // Creates a transaction that sends the specified value from the given account to
+    // the provided recipient address, using a greedy input selector and the default
+    // mutli-output change strategy.
+    pub fn create_standard_transaction(
+        &mut self,
+        from_account: &TestAccount<DbT::Account>,
+        to: ZcashAddress,
+        value: NonNegativeAmount,
+    ) -> Result<
+        NonEmpty<TxId>,
+        super::wallet::TransferErrT<
+            DbT,
+            GreedyInputSelector<DbT>,
+            standard::MultiOutputChangeStrategy<DbT>,
+        >,
+    > {
+        let input_selector = GreedyInputSelector::new();
+
+        #[cfg(not(feature = "orchard"))]
+        let fallback_change_pool = ShieldedProtocol::Sapling;
+        #[cfg(feature = "orchard")]
+        let fallback_change_pool = ShieldedProtocol::Orchard;
+
+        let change_strategy = standard::SingleOutputChangeStrategy::new(
+            StandardFeeRule::Zip317,
+            None,
+            fallback_change_pool,
+            DustOutputPolicy::default(),
+        );
+
+        let request =
+            zip321::TransactionRequest::new(vec![Payment::without_memo(to, value)]).unwrap();
+
+        self.spend(
+            &input_selector,
+            &change_strategy,
+            from_account.usk(),
+            request,
+            OvkPolicy::Sender,
+            NonZeroU32::MIN,
+        )
+    }
+
     /// Prepares and executes the given [`zip321::TransactionRequest`] in a single step.
     #[allow(clippy::type_complexity)]
     pub fn spend<InputsT, ChangeT>(
@@ -2351,12 +2396,12 @@ impl InputSource for MockWalletDb {
         Ok(SpendableNotes::empty())
     }
 
-    fn get_wallet_metadata(
+    fn get_account_metadata(
         &self,
         _account: Self::AccountId,
-        _min_value: NonNegativeAmount,
+        _selector: &NoteFilter,
         _exclude: &[Self::NoteRef],
-    ) -> Result<WalletMeta, Self::Error> {
+    ) -> Result<AccountMeta, Self::Error> {
         Err(())
     }
 }
diff --git a/zcash_client_backend/src/data_api/testing/pool.rs b/zcash_client_backend/src/data_api/testing/pool.rs
index 846a68113..756be223a 100644
--- a/zcash_client_backend/src/data_api/testing/pool.rs
+++ b/zcash_client_backend/src/data_api/testing/pool.rs
@@ -2,7 +2,7 @@ use std::{
     cmp::Eq,
     convert::Infallible,
     hash::Hash,
-    num::{NonZeroU32, NonZeroU8, NonZeroUsize},
+    num::{NonZeroU32, NonZeroU64, NonZeroU8, NonZeroUsize},
 };
 
 use assert_matches::assert_matches;
@@ -43,8 +43,8 @@ use crate::{
         wallet::{
             decrypt_and_store_transaction, input_selection::GreedyInputSelector, TransferErrT,
         },
-        Account as _, AccountBirthday, DecryptedTransaction, InputSource, Ratio,
-        WalletCommitmentTrees, WalletRead, WalletSummary, WalletTest, WalletWrite,
+        Account as _, AccountBirthday, BoundedU8, DecryptedTransaction, InputSource, NoteFilter,
+        Ratio, WalletCommitmentTrees, WalletRead, WalletSummary, WalletTest, WalletWrite,
     },
     decrypt_transaction,
     fees::{
@@ -362,7 +362,7 @@ pub fn send_with_multiple_change_outputs<T: ShieldedPoolTester>(
         Some(change_memo.clone().into()),
         T::SHIELDED_PROTOCOL,
         DustOutputPolicy::default(),
-        SplitPolicy::new(
+        SplitPolicy::with_min_output_value(
             NonZeroUsize::new(2).unwrap(),
             NonNegativeAmount::const_from_u64(100_0000),
         ),
@@ -465,7 +465,7 @@ pub fn send_with_multiple_change_outputs<T: ShieldedPoolTester>(
         Some(change_memo.into()),
         T::SHIELDED_PROTOCOL,
         DustOutputPolicy::default(),
-        SplitPolicy::new(
+        SplitPolicy::with_min_output_value(
             NonZeroUsize::new(8).unwrap(),
             NonNegativeAmount::const_from_u64(10_0000),
         ),
@@ -530,7 +530,7 @@ pub fn send_multi_step_proposed_transfer<T: ShieldedPoolTester, DSF>(
         // Add funds to the wallet.
         add_funds(st, value);
 
-        let expected_step0_fee = (zip317::MARGINAL_FEE * 3).unwrap();
+        let expected_step0_fee = (zip317::MARGINAL_FEE * 3u64).unwrap();
         let expected_step1_fee = zip317::MINIMUM_FEE;
         let expected_ephemeral = (transfer_amount + expected_step1_fee).unwrap();
         let expected_step0_change =
@@ -1123,7 +1123,7 @@ pub fn spend_fails_on_unverified_notes<T: ShieldedPoolTester>(
     st.scan_cached_blocks(h2 + 1, 8);
 
     // Total balance is value * number of blocks scanned (10).
-    assert_eq!(st.get_total_balance(account_id), (value * 10).unwrap());
+    assert_eq!(st.get_total_balance(account_id), (value * 10u64).unwrap());
 
     // Spend still fails
     assert_matches!(
@@ -1150,15 +1150,15 @@ pub fn spend_fails_on_unverified_notes<T: ShieldedPoolTester>(
     st.scan_cached_blocks(h11, 1);
 
     // Total balance is value * number of blocks scanned (11).
-    assert_eq!(st.get_total_balance(account_id), (value * 11).unwrap());
+    assert_eq!(st.get_total_balance(account_id), (value * 11u64).unwrap());
     // Spendable balance at 10 confirmations is value * 2.
     assert_eq!(
         st.get_spendable_balance(account_id, 10),
-        (value * 2).unwrap()
+        (value * 2u64).unwrap()
     );
     assert_eq!(
         st.get_pending_shielded_balance(account_id, 10),
-        (value * 9).unwrap()
+        (value * 9u64).unwrap()
     );
 
     // Should now be able to generate a proposal
@@ -1192,7 +1192,7 @@ pub fn spend_fails_on_unverified_notes<T: ShieldedPoolTester>(
     // TODO: send to an account so that we can check its balance.
     assert_eq!(
         st.get_total_balance(account_id),
-        ((value * 11).unwrap()
+        ((value * 11u64).unwrap()
             - (amount_sent + NonNegativeAmount::from_u64(10000).unwrap()).unwrap())
         .unwrap()
     );
@@ -2124,7 +2124,7 @@ pub fn fully_funded_fully_private<P0: ShieldedPoolTester, P1: ShieldedPoolTester
     st.generate_next_block(&p1_fvk, AddressType::DefaultExternal, note_value);
     st.scan_cached_blocks(account.birthday().height(), 2);
 
-    let initial_balance = (note_value * 2).unwrap();
+    let initial_balance = (note_value * 2u64).unwrap();
     assert_eq!(st.get_total_balance(account.id()), initial_balance);
     assert_eq!(st.get_spendable_balance(account.id(), 1), initial_balance);
 
@@ -2214,7 +2214,7 @@ pub fn fully_funded_send_to_t<P0: ShieldedPoolTester, P1: ShieldedPoolTester>(
     st.generate_next_block(&p1_fvk, AddressType::DefaultExternal, note_value);
     st.scan_cached_blocks(account.birthday().height(), 2);
 
-    let initial_balance = (note_value * 2).unwrap();
+    let initial_balance = (note_value * 2u64).unwrap();
     assert_eq!(st.get_total_balance(account.id()), initial_balance);
     assert_eq!(st.get_spendable_balance(account.id(), 1), initial_balance);
 
@@ -2307,7 +2307,7 @@ pub fn multi_pool_checkpoint<P0: ShieldedPoolTester, P1: ShieldedPoolTester>(
 
     let next_to_scan = scanned.scanned_range().end;
 
-    let initial_balance = (note_value * 3).unwrap();
+    let initial_balance = (note_value * 3u64).unwrap();
     assert_eq!(st.get_total_balance(acct_id), initial_balance);
     assert_eq!(st.get_spendable_balance(acct_id, 1), initial_balance);
 
@@ -2352,7 +2352,7 @@ pub fn multi_pool_checkpoint<P0: ShieldedPoolTester, P1: ShieldedPoolTester>(
     let expected_change = (note_value - transfer_amount - expected_fee).unwrap();
     assert_eq!(
         st.get_total_balance(acct_id),
-        ((note_value * 2).unwrap() + expected_change).unwrap()
+        ((note_value * 2u64).unwrap() + expected_change).unwrap()
     );
     assert_eq!(st.get_pending_change(acct_id, 1), expected_change);
 
@@ -2396,8 +2396,8 @@ pub fn multi_pool_checkpoint<P0: ShieldedPoolTester, P1: ShieldedPoolTester>(
     );
 
     let expected_final = (initial_balance + note_value
-        - (transfer_amount * 3).unwrap()
-        - (expected_fee * 3).unwrap())
+        - (transfer_amount * 3u64).unwrap()
+        - (expected_fee * 3u64).unwrap())
     .unwrap();
     assert_eq!(st.get_total_balance(acct_id), expected_final);
 
@@ -2972,3 +2972,89 @@ pub fn scan_cached_blocks_detects_spends_out_of_order<T: ShieldedPoolTester, DSF
         (value - value2).unwrap()
     );
 }
+
+pub fn metadata_queries_exclude_unwanted_notes<T: ShieldedPoolTester, DSF, TC>(
+    ds_factory: DSF,
+    cache: TC,
+) where
+    DSF: DataStoreFactory,
+    <DSF as DataStoreFactory>::AccountId: std::fmt::Debug,
+    TC: TestCache,
+{
+    let mut st = TestBuilder::new()
+        .with_data_store_factory(ds_factory)
+        .with_block_cache(cache)
+        .with_account_from_sapling_activation(BlockHash([0; 32]))
+        .build();
+
+    let account = st.test_account().cloned().unwrap();
+    let dfvk = T::test_account_fvk(&st);
+
+    // Create 10 blocks with successively increasing value
+    let value = NonNegativeAmount::const_from_u64(100_0000);
+    let (h0, _, _) = st.generate_next_block(&dfvk, AddressType::DefaultExternal, value);
+    let mut note_values = vec![value];
+    for i in 2..=10 {
+        let value = NonNegativeAmount::const_from_u64(i * 100_0000);
+        st.generate_next_block(&dfvk, AddressType::DefaultExternal, value);
+        note_values.push(value);
+    }
+    st.scan_cached_blocks(h0, 10);
+
+    let test_meta = |st: &TestState<TC, DSF::DataStore, LocalNetwork>, query, expected_count| {
+        let metadata = st
+            .wallet()
+            .get_account_metadata(account.id(), &query, &[])
+            .unwrap();
+
+        assert_eq!(metadata.note_count(T::SHIELDED_PROTOCOL), expected_count);
+    };
+
+    test_meta(
+        &st,
+        NoteFilter::ExceedsMinValue(NonNegativeAmount::const_from_u64(1000_0000)),
+        Some(1),
+    );
+    test_meta(
+        &st,
+        NoteFilter::ExceedsMinValue(NonNegativeAmount::const_from_u64(500_0000)),
+        Some(6),
+    );
+    test_meta(
+        &st,
+        NoteFilter::ExceedsBalancePercentage(BoundedU8::new_const(10)),
+        Some(5),
+    );
+
+    // We haven't sent any funds yet, so we can't evaluate this query
+    test_meta(
+        &st,
+        NoteFilter::ExceedsPriorSendPercentile(BoundedU8::new_const(50)),
+        None,
+    );
+
+    // Spend half of each one of our notes, so that we can get a distribution of sent note values.
+    // FIXME: This test is currently excessively specialized to the `zcash_client_sqlite::WalletDb`
+    // implmentation of the `InputSource` trait. A better approach would be to create a test input
+    // source that can select a set of notes directly based upon their nullifiers.
+    let not_our_key = T::sk_to_fvk(&T::sk(&[0xf5; 32]));
+    let to = T::fvk_default_address(&not_our_key).to_zcash_address(st.network());
+    let nz2 = NonZeroU64::new(2).unwrap();
+
+    for value in &note_values {
+        let txids = st
+            .create_standard_transaction(&account, to.clone(), *value / nz2)
+            .unwrap();
+        st.generate_next_block_including(txids.head);
+    }
+    st.scan_cached_blocks(h0 + 10, 10);
+
+    // Since we've spent half our notes, our remaining notes each have approximately half their
+    // original value. The 50th percentile of our spends should be 250_0000 ZAT, and half of our
+    // remaining notes should have value greater than that.
+    test_meta(
+        &st,
+        NoteFilter::ExceedsPriorSendPercentile(BoundedU8::new_const(50)),
+        Some(5),
+    );
+}
diff --git a/zcash_client_backend/src/fees.rs b/zcash_client_backend/src/fees.rs
index a10136db5..eba11c7e6 100644
--- a/zcash_client_backend/src/fees.rs
+++ b/zcash_client_backend/src/fees.rs
@@ -10,7 +10,8 @@ use zcash_primitives::{
         components::{amount::NonNegativeAmount, OutPoint},
         fees::{
             transparent::{self, InputSize},
-            zip317 as prim_zip317, FeeRule,
+            zip317::{self as prim_zip317},
+            FeeRule,
         },
     },
 };
@@ -352,21 +353,32 @@ impl Default for DustOutputPolicy {
 
 /// A policy that describes how change output should be split into multiple notes for the purpose
 /// of note management.
+///
+/// If an account contains at least [`Self::target_output_count`] notes having at least value
+/// [`Self::min_split_output_value`], this policy will recommend a single output; if the account
+/// contains fewer such notes, this policy will recommend that multiple outputs be produced in
+/// order to achieve the target.
 #[derive(Clone, Copy, Debug)]
 pub struct SplitPolicy {
     target_output_count: NonZeroUsize,
-    min_split_output_size: NonNegativeAmount,
+    min_split_output_value: Option<NonNegativeAmount>,
 }
 
 impl SplitPolicy {
-    /// Constructs a new [`SplitPolicy`] from its constituent parts.
-    pub fn new(
+    /// In the case that no other conditions provided by the user are available to fall back on,
+    /// a default value of [`MARGINAL_FEE`] * 100 will be used as the "minimum usable note value"
+    /// when retrieving wallet metadata.
+    pub(crate) const MIN_NOTE_VALUE: NonNegativeAmount = NonNegativeAmount::const_from_u64(500000);
+
+    /// Constructs a new [`SplitPolicy`] that splits change to ensure the given number of spendable
+    /// outputs exists within an account, each having at least the specified minimum note value.
+    pub fn with_min_output_value(
         target_output_count: NonZeroUsize,
-        min_split_output_size: NonNegativeAmount,
+        min_split_output_value: NonNegativeAmount,
     ) -> Self {
         Self {
             target_output_count,
-            min_split_output_size,
+            min_split_output_value: Some(min_split_output_value),
         }
     }
 
@@ -374,44 +386,71 @@ impl SplitPolicy {
     pub fn single_output() -> Self {
         Self {
             target_output_count: NonZeroUsize::MIN,
-            min_split_output_size: NonNegativeAmount::ZERO,
+            min_split_output_value: None,
         }
     }
 
+    /// Returns the number of outputs that this policy will attempt to ensure that the wallet has
+    /// available for spending.
+    pub fn target_output_count(&self) -> NonZeroUsize {
+        self.target_output_count
+    }
+
     /// Returns the minimum value for a note resulting from splitting of change.
-    ///
-    /// If splitting change would result in notes of value less than the minimum split output size,
-    /// a smaller number of splits should be chosen.
-    pub fn min_split_output_size(&self) -> NonNegativeAmount {
-        self.min_split_output_size
+    pub fn min_split_output_value(&self) -> Option<NonNegativeAmount> {
+        self.min_split_output_value
     }
 
     /// Returns the number of output notes to produce from the given total change value, given the
-    /// number of existing unspent notes in the account and this policy.
+    /// total value and number of existing unspent notes in the account and this policy.
+    ///
+    /// If splitting change to produce [`Self::target_output_count`] would result in notes of value less than
+    /// [`Self::min_split_output_value`], then this will produce
+    ///
     pub fn split_count(
         &self,
-        existing_notes: usize,
+        existing_notes: Option<usize>,
+        existing_notes_total: Option<NonNegativeAmount>,
         total_change: NonNegativeAmount,
     ) -> NonZeroUsize {
-        let mut split_count =
-            NonZeroUsize::new(usize::from(self.target_output_count).saturating_sub(existing_notes))
-                .unwrap_or(NonZeroUsize::MIN);
-
-        loop {
-            let per_output_change = total_change.div_with_remainder(
-                NonZeroU64::new(
-                    u64::try_from(usize::from(split_count)).expect("usize fits into u64"),
-                )
-                .unwrap(),
-            );
-            if *per_output_change.quotient() >= self.min_split_output_size {
-                return split_count;
-            } else if let Some(new_count) = NonZeroUsize::new(usize::from(split_count) - 1) {
-                split_count = new_count;
-            } else {
-                // We always create at least one change output.
-                return NonZeroUsize::MIN;
+        fn to_nonzero_u64(value: usize) -> NonZeroU64 {
+            NonZeroU64::new(u64::try_from(value).expect("usize fits into u64"))
+                .expect("NonZeroU64 input derived from NonZeroUsize")
+        }
+
+        let mut split_count = NonZeroUsize::new(
+            usize::from(self.target_output_count)
+                .saturating_sub(existing_notes.unwrap_or(usize::MAX)),
+        )
+        .unwrap_or(NonZeroUsize::MIN);
+
+        let min_split_output_value = self.min_split_output_value.or_else(|| {
+            // If no minimum split output size is set, we choose the minimum split size to be a
+            // quarter of the average value of notes in the wallet after the transaction.
+            (existing_notes_total + total_change).map(|total| {
+                *total
+                    .div_with_remainder(to_nonzero_u64(
+                        usize::from(self.target_output_count).saturating_mul(4),
+                    ))
+                    .quotient()
+            })
+        });
+
+        if let Some(min_split_output_value) = min_split_output_value {
+            loop {
+                let per_output_change =
+                    total_change.div_with_remainder(to_nonzero_u64(usize::from(split_count)));
+                if *per_output_change.quotient() >= min_split_output_value {
+                    return split_count;
+                } else if let Some(new_count) = NonZeroUsize::new(usize::from(split_count) - 1) {
+                    split_count = new_count;
+                } else {
+                    // We always create at least one change output.
+                    return NonZeroUsize::MIN;
+                }
             }
+        } else {
+            NonZeroUsize::MIN
         }
     }
 }
@@ -466,7 +505,7 @@ pub trait ChangeStrategy {
 
     /// Tye type of wallet metadata that this change strategy relies upon in order to compute
     /// change.
-    type WalletMetaT;
+    type AccountMetaT;
 
     /// Returns the fee rule that this change strategy will respect when performing
     /// balance computations.
@@ -479,7 +518,7 @@ pub trait ChangeStrategy {
         meta_source: &Self::MetaSource,
         account: <Self::MetaSource as InputSource>::AccountId,
         exclude: &[<Self::MetaSource as InputSource>::NoteRef],
-    ) -> Result<Self::WalletMetaT, <Self::MetaSource as InputSource>::Error>;
+    ) -> Result<Self::AccountMetaT, <Self::MetaSource as InputSource>::Error>;
 
     /// Computes the totals of inputs, suggested change amounts, and fees given the
     /// provided inputs and outputs being used to construct a transaction.
@@ -516,7 +555,7 @@ pub trait ChangeStrategy {
         sapling: &impl sapling::BundleView<NoteRefT>,
         #[cfg(feature = "orchard")] orchard: &impl orchard::BundleView<NoteRefT>,
         ephemeral_balance: Option<&EphemeralBalance>,
-        wallet_meta: &Self::WalletMetaT,
+        wallet_meta: &Self::AccountMetaT,
     ) -> Result<TransactionBalance, ChangeError<Self::Error, NoteRefT>>;
 }
 
diff --git a/zcash_client_backend/src/fees/common.rs b/zcash_client_backend/src/fees/common.rs
index 2a1a4a4e2..8aa7f38f4 100644
--- a/zcash_client_backend/src/fees/common.rs
+++ b/zcash_client_backend/src/fees/common.rs
@@ -11,7 +11,7 @@ use zcash_primitives::{
 };
 use zcash_protocol::ShieldedProtocol;
 
-use crate::data_api::WalletMeta;
+use crate::data_api::AccountMeta;
 
 use super::{
     sapling as sapling_fees, ChangeError, ChangeValue, DustAction, DustOutputPolicy,
@@ -203,7 +203,7 @@ impl<'a, P, F> SinglePoolBalanceConfig<'a, P, F> {
 #[allow(clippy::too_many_arguments)]
 pub(crate) fn single_pool_output_balance<P: consensus::Parameters, NoteRefT: Clone, F: FeeRule, E>(
     cfg: SinglePoolBalanceConfig<P, F>,
-    wallet_meta: Option<&WalletMeta>,
+    wallet_meta: Option<&AccountMeta>,
     target_height: BlockHeight,
     transparent_inputs: &[impl transparent::InputView],
     transparent_outputs: &[impl transparent::OutputView],
@@ -235,7 +235,8 @@ where
     let change_pool = select_change_pool(&net_flows, cfg.fallback_change_pool);
     let target_change_count = wallet_meta.map_or(1, |m| {
         usize::from(cfg.split_policy.target_output_count)
-            .saturating_sub(m.total_note_count())
+            // If we cannot determine a total note count, fall back to a single output
+            .saturating_sub(m.total_note_count().unwrap_or(usize::MAX))
             .max(1)
     });
     let target_change_counts = OutputManifest {
@@ -429,8 +430,11 @@ where
             // available in the wallet, irrespective of pool. If we don't have any wallet metadata
             // available, we fall back to generating a single change output.
             let split_count = wallet_meta.map_or(NonZeroUsize::MIN, |wm| {
-                cfg.split_policy
-                    .split_count(wm.total_note_count(), proposed_change)
+                cfg.split_policy.split_count(
+                    wm.total_note_count(),
+                    wm.total_value(),
+                    proposed_change,
+                )
             });
             let per_output_change = proposed_change.div_with_remainder(
                 NonZeroU64::new(
@@ -531,8 +535,8 @@ where
                         // We can add a change output if necessary.
                         assert!(fee_with_change <= fee_with_dust);
 
-                        let reasonable_fee =
-                            (fee_with_change + (MINIMUM_FEE * 10).unwrap()).ok_or_else(overflow)?;
+                        let reasonable_fee = (fee_with_change + (MINIMUM_FEE * 10u64).unwrap())
+                            .ok_or_else(overflow)?;
 
                         if fee_with_dust > reasonable_fee {
                             // Defend against losing money by using AddDustToFee with a too-high
diff --git a/zcash_client_backend/src/fees/fixed.rs b/zcash_client_backend/src/fees/fixed.rs
index 49cd86114..b755d653b 100644
--- a/zcash_client_backend/src/fees/fixed.rs
+++ b/zcash_client_backend/src/fees/fixed.rs
@@ -60,7 +60,7 @@ impl<I: InputSource> ChangeStrategy for SingleOutputChangeStrategy<I> {
     type FeeRule = FixedFeeRule;
     type Error = BalanceError;
     type MetaSource = I;
-    type WalletMetaT = ();
+    type AccountMetaT = ();
 
     fn fee_rule(&self) -> &Self::FeeRule {
         &self.fee_rule
@@ -71,7 +71,7 @@ impl<I: InputSource> ChangeStrategy for SingleOutputChangeStrategy<I> {
         _meta_source: &Self::MetaSource,
         _account: <Self::MetaSource as InputSource>::AccountId,
         _exclude: &[<Self::MetaSource as crate::data_api::InputSource>::NoteRef],
-    ) -> Result<Self::WalletMetaT, <Self::MetaSource as crate::data_api::InputSource>::Error> {
+    ) -> Result<Self::AccountMetaT, <Self::MetaSource as crate::data_api::InputSource>::Error> {
         Ok(())
     }
 
@@ -84,7 +84,7 @@ impl<I: InputSource> ChangeStrategy for SingleOutputChangeStrategy<I> {
         sapling: &impl sapling_fees::BundleView<NoteRefT>,
         #[cfg(feature = "orchard")] orchard: &impl orchard_fees::BundleView<NoteRefT>,
         ephemeral_balance: Option<&EphemeralBalance>,
-        _wallet_meta: &Self::WalletMetaT,
+        _wallet_meta: &Self::AccountMetaT,
     ) -> Result<TransactionBalance, ChangeError<Self::Error, NoteRefT>> {
         let split_policy = SplitPolicy::single_output();
         let cfg = SinglePoolBalanceConfig::new(
diff --git a/zcash_client_backend/src/fees/zip317.rs b/zcash_client_backend/src/fees/zip317.rs
index 7d89897be..e654bb56d 100644
--- a/zcash_client_backend/src/fees/zip317.rs
+++ b/zcash_client_backend/src/fees/zip317.rs
@@ -14,7 +14,7 @@ use zcash_primitives::{
 use zcash_protocol::value::{BalanceError, Zatoshis};
 
 use crate::{
-    data_api::{InputSource, WalletMeta},
+    data_api::{AccountMeta, InputSource, NoteFilter},
     fees::StandardFeeRule,
     ShieldedProtocol,
 };
@@ -101,7 +101,7 @@ where
     type FeeRule = R;
     type Error = <R as FeeRule>::Error;
     type MetaSource = I;
-    type WalletMetaT = ();
+    type AccountMetaT = ();
 
     fn fee_rule(&self) -> &Self::FeeRule {
         &self.fee_rule
@@ -112,7 +112,7 @@ where
         _meta_source: &Self::MetaSource,
         _account: <Self::MetaSource as InputSource>::AccountId,
         _exclude: &[<Self::MetaSource as InputSource>::NoteRef],
-    ) -> Result<Self::WalletMetaT, <Self::MetaSource as InputSource>::Error> {
+    ) -> Result<Self::AccountMetaT, <Self::MetaSource as InputSource>::Error> {
         Ok(())
     }
 
@@ -125,7 +125,7 @@ where
         sapling: &impl sapling_fees::BundleView<NoteRefT>,
         #[cfg(feature = "orchard")] orchard: &impl orchard_fees::BundleView<NoteRefT>,
         ephemeral_balance: Option<&EphemeralBalance>,
-        _wallet_meta: &Self::WalletMetaT,
+        _wallet_meta: &Self::AccountMetaT,
     ) -> Result<TransactionBalance, ChangeError<Self::Error, NoteRefT>> {
         let split_policy = SplitPolicy::single_output();
         let cfg = SinglePoolBalanceConfig::new(
@@ -204,7 +204,7 @@ where
     type FeeRule = R;
     type Error = <R as FeeRule>::Error;
     type MetaSource = I;
-    type WalletMetaT = WalletMeta;
+    type AccountMetaT = AccountMeta;
 
     fn fee_rule(&self) -> &Self::FeeRule {
         &self.fee_rule
@@ -215,8 +215,14 @@ where
         meta_source: &Self::MetaSource,
         account: <Self::MetaSource as InputSource>::AccountId,
         exclude: &[<Self::MetaSource as InputSource>::NoteRef],
-    ) -> Result<Self::WalletMetaT, <Self::MetaSource as InputSource>::Error> {
-        meta_source.get_wallet_metadata(account, self.split_policy.min_split_output_size(), exclude)
+    ) -> Result<Self::AccountMetaT, <Self::MetaSource as InputSource>::Error> {
+        let note_selector = NoteFilter::ExceedsMinValue(
+            self.split_policy
+                .min_split_output_value()
+                .unwrap_or(SplitPolicy::MIN_NOTE_VALUE),
+        );
+
+        meta_source.get_account_metadata(account, &note_selector, exclude)
     }
 
     fn compute_balance<P: consensus::Parameters, NoteRefT: Clone>(
@@ -228,7 +234,7 @@ where
         sapling: &impl sapling_fees::BundleView<NoteRefT>,
         #[cfg(feature = "orchard")] orchard: &impl orchard_fees::BundleView<NoteRefT>,
         ephemeral_balance: Option<&EphemeralBalance>,
-        wallet_meta: &Self::WalletMetaT,
+        wallet_meta: &Self::AccountMetaT,
     ) -> Result<TransactionBalance, ChangeError<Self::Error, NoteRefT>> {
         let cfg = SinglePoolBalanceConfig::new(
             params,
@@ -271,7 +277,9 @@ mod tests {
 
     use super::SingleOutputChangeStrategy;
     use crate::{
-        data_api::{testing::MockWalletDb, wallet::input_selection::SaplingPayment, WalletMeta},
+        data_api::{
+            testing::MockWalletDb, wallet::input_selection::SaplingPayment, AccountMeta, PoolMeta,
+        },
         fees::{
             tests::{TestSaplingInput, TestTransparentInput},
             zip317::MultiOutputChangeStrategy,
@@ -334,7 +342,7 @@ mod tests {
             None,
             ShieldedProtocol::Sapling,
             DustOutputPolicy::default(),
-            SplitPolicy::new(
+            SplitPolicy::with_min_output_value(
                 NonZeroUsize::new(5).unwrap(),
                 NonNegativeAmount::const_from_u64(100_0000),
             ),
@@ -342,7 +350,7 @@ mod tests {
 
         {
             // spend a single Sapling note and produce 5 outputs
-            let balance = |existing_notes| {
+            let balance = |existing_notes, total| {
                 change_strategy.compute_balance(
                     &Network::TestNetwork,
                     Network::TestNetwork
@@ -363,16 +371,12 @@ mod tests {
                     #[cfg(feature = "orchard")]
                     &orchard_fees::EmptyBundleView,
                     None,
-                    &WalletMeta::new(
-                        existing_notes,
-                        #[cfg(feature = "orchard")]
-                        0,
-                    ),
+                    &AccountMeta::new(Some(PoolMeta::new(existing_notes, total)), None),
                 )
             };
 
             assert_matches!(
-                balance(0),
+                balance(0, NonNegativeAmount::ZERO),
                 Ok(balance) if
                     balance.proposed_change() == [
                         ChangeValue::sapling(NonNegativeAmount::const_from_u64(129_4000), None),
@@ -385,7 +389,7 @@ mod tests {
             );
 
             assert_matches!(
-                balance(2),
+                balance(2, NonNegativeAmount::const_from_u64(100_0000)),
                 Ok(balance) if
                     balance.proposed_change() == [
                         ChangeValue::sapling(NonNegativeAmount::const_from_u64(216_0000), None),
@@ -419,10 +423,9 @@ mod tests {
                 #[cfg(feature = "orchard")]
                 &orchard_fees::EmptyBundleView,
                 None,
-                &WalletMeta::new(
-                    0,
-                    #[cfg(feature = "orchard")]
-                    0,
+                &AccountMeta::new(
+                    Some(PoolMeta::new(0, NonNegativeAmount::ZERO)),
+                    Some(PoolMeta::new(0, NonNegativeAmount::ZERO)),
                 ),
             );
 
diff --git a/zcash_client_sqlite/CHANGELOG.md b/zcash_client_sqlite/CHANGELOG.md
index 035313f54..0851eb471 100644
--- a/zcash_client_sqlite/CHANGELOG.md
+++ b/zcash_client_sqlite/CHANGELOG.md
@@ -15,6 +15,7 @@ and this library adheres to Rust's notion of
 - MSRV is now 1.77.0.
 - Migrated from `schemer` to our fork `schemerz`.
 - Migrated to `rusqlite 0.32`.
+- `error::SqliteClientError` has additional variant `NoteSelectorInvalid`
 
 ## [0.12.2] - 2024-10-21
 
diff --git a/zcash_client_sqlite/src/error.rs b/zcash_client_sqlite/src/error.rs
index 741003b6e..43be98939 100644
--- a/zcash_client_sqlite/src/error.rs
+++ b/zcash_client_sqlite/src/error.rs
@@ -5,6 +5,7 @@ use std::fmt;
 
 use shardtree::error::ShardTreeError;
 use zcash_address::ParseError;
+use zcash_client_backend::data_api::NoteFilter;
 use zcash_client_backend::PoolType;
 use zcash_keys::keys::AddressGenerationError;
 use zcash_primitives::zip32;
@@ -121,6 +122,9 @@ pub enum SqliteClientError {
     /// An error occurred in computing wallet balance
     BalanceError(BalanceError),
 
+    /// A note selection query contained an invalid constant or was otherwise not supported.
+    NoteFilterInvalid(NoteFilter),
+
     /// The proposal cannot be constructed until transactions with previously reserved
     /// ephemeral address outputs have been mined. The parameters are the account id and
     /// the index that could not safely be reserved.
@@ -187,6 +191,7 @@ impl fmt::Display for SqliteClientError {
             SqliteClientError::ChainHeightUnknown => write!(f, "Chain height unknown; please call `update_chain_tip`"),
             SqliteClientError::UnsupportedPoolType(t) => write!(f, "Pool type is not currently supported: {}", t),
             SqliteClientError::BalanceError(e) => write!(f, "Balance error: {}", e),
+            SqliteClientError::NoteFilterInvalid(s) => write!(f, "Could not evaluate selection query: {:?}", s),
             #[cfg(feature = "transparent-inputs")]
             SqliteClientError::ReachedGapLimit(account_id, bad_index) => write!(f,
                 "The proposal cannot be constructed until transactions with previously reserved ephemeral address outputs have been mined. \
diff --git a/zcash_client_sqlite/src/lib.rs b/zcash_client_sqlite/src/lib.rs
index f11022927..74e2d23a4 100644
--- a/zcash_client_sqlite/src/lib.rs
+++ b/zcash_client_sqlite/src/lib.rs
@@ -50,10 +50,10 @@ use zcash_client_backend::{
         self,
         chain::{BlockSource, ChainState, CommitmentTreeRoot},
         scanning::{ScanPriority, ScanRange},
-        Account, AccountBirthday, AccountPurpose, AccountSource, BlockMetadata,
-        DecryptedTransaction, InputSource, NullifierQuery, ScannedBlock, SeedRelevance,
-        SentTransaction, SpendableNotes, TransactionDataRequest, WalletCommitmentTrees, WalletMeta,
-        WalletRead, WalletSummary, WalletWrite, SAPLING_SHARD_HEIGHT,
+        Account, AccountBirthday, AccountMeta, AccountPurpose, AccountSource, BlockMetadata,
+        DecryptedTransaction, InputSource, NoteFilter, NullifierQuery, ScannedBlock, SeedRelevance,
+        SentTransaction, SpendableNotes, TransactionDataRequest, WalletCommitmentTrees, WalletRead,
+        WalletSummary, WalletWrite, SAPLING_SHARD_HEIGHT,
     },
     keys::{
         AddressGenerationError, UnifiedAddressRequest, UnifiedFullViewingKey, UnifiedSpendingKey,
@@ -128,7 +128,7 @@ pub mod error;
 pub mod wallet;
 use wallet::{
     commitment_tree::{self, put_shard_roots},
-    common::count_outputs,
+    common::spendable_notes_meta,
     SubtreeProgressEstimator,
 };
 
@@ -348,39 +348,38 @@ impl<C: Borrow<rusqlite::Connection>, P: consensus::Parameters> InputSource for
         )
     }
 
-    fn get_wallet_metadata(
+    /// Returns metadata for the spendable notes in the wallet.
+    fn get_account_metadata(
         &self,
         account_id: Self::AccountId,
-        min_value: NonNegativeAmount,
+        selector: &NoteFilter,
         exclude: &[Self::NoteRef],
-    ) -> Result<WalletMeta, Self::Error> {
+    ) -> Result<AccountMeta, Self::Error> {
         let chain_tip_height = wallet::chain_tip_height(self.conn.borrow())?
             .ok_or(SqliteClientError::ChainHeightUnknown)?;
 
-        let sapling_note_count = count_outputs(
+        let sapling_pool_meta = spendable_notes_meta(
             self.conn.borrow(),
-            account_id,
-            min_value,
-            exclude,
             ShieldedProtocol::Sapling,
             chain_tip_height,
+            account_id,
+            selector,
+            exclude,
         )?;
 
         #[cfg(feature = "orchard")]
-        let orchard_note_count = count_outputs(
+        let orchard_pool_meta = spendable_notes_meta(
             self.conn.borrow(),
-            account_id,
-            min_value,
-            exclude,
             ShieldedProtocol::Orchard,
             chain_tip_height,
+            account_id,
+            selector,
+            exclude,
         )?;
+        #[cfg(not(feature = "orchard"))]
+        let orchard_pool_meta = None;
 
-        Ok(WalletMeta::new(
-            sapling_note_count,
-            #[cfg(feature = "orchard")]
-            orchard_note_count,
-        ))
+        Ok(AccountMeta::new(sapling_pool_meta, orchard_pool_meta))
     }
 }
 
diff --git a/zcash_client_sqlite/src/testing/pool.rs b/zcash_client_sqlite/src/testing/pool.rs
index 8465a582a..726a37703 100644
--- a/zcash_client_sqlite/src/testing/pool.rs
+++ b/zcash_client_sqlite/src/testing/pool.rs
@@ -254,3 +254,10 @@ pub(crate) fn scan_cached_blocks_detects_spends_out_of_order<T: ShieldedPoolTest
         _,
     >(TestDbFactory::default(), BlockCache::new())
 }
+
+pub(crate) fn metadata_queries_exclude_unwanted_notes<T: ShieldedPoolTester>() {
+    zcash_client_backend::data_api::testing::pool::metadata_queries_exclude_unwanted_notes::<T, _, _>(
+        TestDbFactory::default(),
+        BlockCache::new(),
+    )
+}
diff --git a/zcash_client_sqlite/src/wallet/common.rs b/zcash_client_sqlite/src/wallet/common.rs
index 378abbe1a..b76b1a2bf 100644
--- a/zcash_client_sqlite/src/wallet/common.rs
+++ b/zcash_client_sqlite/src/wallet/common.rs
@@ -1,14 +1,24 @@
 //! Functions common to Sapling and Orchard support in the wallet.
 
 use rusqlite::{named_params, types::Value, Connection, Row};
-use std::rc::Rc;
+use std::{num::NonZeroU64, rc::Rc};
 
-use zcash_client_backend::{wallet::ReceivedNote, ShieldedProtocol};
+use zcash_client_backend::{
+    data_api::{NoteFilter, PoolMeta},
+    wallet::ReceivedNote,
+    ShieldedProtocol,
+};
 use zcash_primitives::transaction::{components::amount::NonNegativeAmount, TxId};
-use zcash_protocol::consensus::{self, BlockHeight};
+use zcash_protocol::{
+    consensus::{self, BlockHeight},
+    value::BalanceError,
+    PoolType,
+};
 
 use super::wallet_birthday;
-use crate::{error::SqliteClientError, AccountId, ReceivedNoteId, SAPLING_TABLES_PREFIX};
+use crate::{
+    error::SqliteClientError, wallet::pool_code, AccountId, ReceivedNoteId, SAPLING_TABLES_PREFIX,
+};
 
 #[cfg(feature = "orchard")]
 use crate::ORCHARD_TABLES_PREFIX;
@@ -226,14 +236,14 @@ where
         .collect::<Result<_, _>>()
 }
 
-pub(crate) fn count_outputs(
+pub(crate) fn spendable_notes_meta(
     conn: &rusqlite::Connection,
-    account: AccountId,
-    min_value: NonNegativeAmount,
-    exclude: &[ReceivedNoteId],
     protocol: ShieldedProtocol,
     chain_tip_height: BlockHeight,
-) -> Result<usize, rusqlite::Error> {
+    account: AccountId,
+    selector: &NoteFilter,
+    exclude: &[ReceivedNoteId],
+) -> Result<Option<PoolMeta>, SqliteClientError> {
     let (table_prefix, _, _) = per_protocol_names(protocol);
 
     let excluded: Vec<Value> = exclude
@@ -248,33 +258,181 @@ pub(crate) fn count_outputs(
         .collect();
     let excluded_ptr = Rc::new(excluded);
 
-    conn.query_row(
-        &format!(
-            "SELECT COUNT(*)
-             FROM {table_prefix}_received_notes rn
-             INNER JOIN accounts ON accounts.id = rn.account_id
-             INNER JOIN transactions ON transactions.id_tx = rn.tx
-             WHERE value >= :min_value
-             AND accounts.id = :account_id
-             AND accounts.ufvk IS NOT NULL
-             AND recipient_key_scope IS NOT NULL
-             AND transactions.mined_height IS NOT NULL
-             AND rn.id NOT IN rarray(:exclude)
-             AND rn.id NOT IN (
-               SELECT {table_prefix}_received_note_id
-               FROM {table_prefix}_received_note_spends
-               JOIN transactions stx ON stx.id_tx = transaction_id
-               WHERE stx.block IS NOT NULL -- the spending tx is mined
-               OR stx.expiry_height IS NULL -- the spending tx will not expire
-               OR stx.expiry_height > :chain_tip_height -- the spending tx is unexpired
-             )"
-        ),
-        named_params![
-            ":account_id": account.0,
-            ":min_value": u64::from(min_value),
-            ":exclude": &excluded_ptr,
-            ":chain_tip_height": u32::from(chain_tip_height)
-        ],
-        |row| row.get(0),
-    )
+    fn zatoshis(value: i64) -> Result<NonNegativeAmount, SqliteClientError> {
+        NonNegativeAmount::from_nonnegative_i64(value).map_err(|_| {
+            SqliteClientError::CorruptedData(format!("Negative received note value: {}", value))
+        })
+    }
+
+    let run_selection = |min_value| {
+        conn.query_row_and_then::<_, SqliteClientError, _, _>(
+            &format!(
+                "SELECT COUNT(*), SUM(rn.value)
+                 FROM {table_prefix}_received_notes rn
+                 INNER JOIN accounts a ON a.id = rn.account_id
+                 INNER JOIN transactions ON transactions.id_tx = rn.tx
+                 WHERE rn.account_id = :account_id
+                 AND a.ufvk IS NOT NULL
+                 AND rn.value >= :min_value
+                 AND transactions.mined_height IS NOT NULL
+                 AND rn.id NOT IN rarray(:exclude)
+                 AND rn.id NOT IN (
+                   SELECT {table_prefix}_received_note_id
+                   FROM {table_prefix}_received_note_spends rns
+                   JOIN transactions stx ON stx.id_tx = rns.transaction_id
+                   WHERE stx.block IS NOT NULL -- the spending tx is mined
+                   OR stx.expiry_height IS NULL -- the spending tx will not expire
+                   OR stx.expiry_height > :chain_tip_height -- the spending tx is unexpired
+                 )"
+            ),
+            named_params![
+                ":account_id": account.0,
+                ":min_value": u64::from(min_value),
+                ":exclude": &excluded_ptr,
+                ":chain_tip_height": u32::from(chain_tip_height)
+            ],
+            |row| {
+                Ok((
+                    row.get::<_, usize>(0)?,
+                    row.get::<_, Option<i64>>(1)?.map(zatoshis).transpose()?,
+                ))
+            },
+        )
+    };
+
+    // Evaluates the provided note filter conditions against the wallet database in order to
+    // determine the minimum value of notes to be produced by note splitting.
+    fn min_note_value(
+        conn: &rusqlite::Connection,
+        table_prefix: &str,
+        account: AccountId,
+        selector: &NoteFilter,
+        chain_tip_height: BlockHeight,
+    ) -> Result<Option<NonNegativeAmount>, SqliteClientError> {
+        match selector {
+            NoteFilter::ExceedsMinValue(v) => Ok(Some(*v)),
+            NoteFilter::ExceedsPriorSendPercentile(n) => {
+                let mut bucket_query = conn.prepare(
+                    "WITH bucketed AS (
+                        SELECT s.value, NTILE(10) OVER (ORDER BY s.value) AS bucket_index
+                        FROM sent_notes s
+                        JOIN transactions t ON s.tx = t.id_tx
+                        WHERE s.from_account_id = :account_id
+                        -- only count mined transactions
+                        AND t.mined_height IS NOT NULL
+                        -- exclude change and account-internal sends
+                        AND (s.to_account_id IS NULL OR s.from_account_id != s.to_account_id)
+                    )
+                    SELECT MAX(value) as value
+                    FROM bucketed
+                    GROUP BY bucket_index
+                    ORDER BY bucket_index",
+                )?;
+
+                let bucket_maxima = bucket_query
+                    .query_and_then::<_, SqliteClientError, _, _>(
+                        named_params![":account_id": account.0],
+                        |row| {
+                            NonNegativeAmount::from_nonnegative_i64(row.get::<_, i64>(0)?).map_err(
+                                |_| {
+                                    SqliteClientError::CorruptedData(format!(
+                                        "Negative received note value: {}",
+                                        n.value()
+                                    ))
+                                },
+                            )
+                        },
+                    )?
+                    .collect::<Result<Vec<_>, _>>()?;
+
+                // Pick a bucket index by scaling the requested percentile to the number of buckets
+                let i = (bucket_maxima.len() * usize::from(*n) / 100).saturating_sub(1);
+                Ok(bucket_maxima.get(i).copied())
+            }
+            NoteFilter::ExceedsBalancePercentage(p) => {
+                let balance = conn.query_row_and_then::<_, SqliteClientError, _, _>(
+                    &format!(
+                        "SELECT SUM(rn.value)
+                         FROM v_received_outputs rn
+                         INNER JOIN accounts a ON a.id = rn.account_id
+                         INNER JOIN transactions ON transactions.id_tx = rn.transaction_id
+                         WHERE rn.account_id = :account_id
+                         AND a.ufvk IS NOT NULL
+                         AND transactions.mined_height IS NOT NULL
+                         AND rn.pool != :transparent_pool
+                         AND rn.id_within_pool_table NOT IN (
+                           SELECT rns.received_output_id
+                           FROM v_received_output_spends rns
+                           JOIN transactions stx ON stx.id_tx = rns.transaction_id
+                           WHERE rns.pool == rn.pool
+                           AND (
+                               stx.block IS NOT NULL -- the spending tx is mined
+                               OR stx.expiry_height IS NULL -- the spending tx will not expire
+                               OR stx.expiry_height > :chain_tip_height -- the spending tx is unexpired
+                           )
+                         )"
+                    ),
+                    named_params![
+                        ":account_id": account.0,
+                        ":chain_tip_height": u32::from(chain_tip_height),
+                        ":transparent_pool": pool_code(PoolType::Transparent)
+                    ],
+                    |row| row.get::<_, Option<i64>>(0)?.map(zatoshis).transpose(),
+                )?;
+
+                Ok(match balance {
+                    None => None,
+                    Some(b) => {
+                        let numerator = (b * u64::from(p.value())).ok_or(BalanceError::Overflow)?;
+                        Some(numerator / NonZeroU64::new(100).expect("Constant is nonzero."))
+                    }
+                })
+            }
+            NoteFilter::Combine(a, b) => {
+                // All the existing note selectors set lower bounds on note value, so the "and"
+                // operation is just taking the maximum of the two lower bounds.
+                let a_min_value =
+                    min_note_value(conn, table_prefix, account, a.as_ref(), chain_tip_height)?;
+                let b_min_value =
+                    min_note_value(conn, table_prefix, account, b.as_ref(), chain_tip_height)?;
+                Ok(a_min_value
+                    .zip(b_min_value)
+                    .map(|(av, bv)| std::cmp::max(av, bv))
+                    .or(a_min_value)
+                    .or(b_min_value))
+            }
+            NoteFilter::Attempt {
+                condition,
+                fallback,
+            } => {
+                let cond = min_note_value(
+                    conn,
+                    table_prefix,
+                    account,
+                    condition.as_ref(),
+                    chain_tip_height,
+                )?;
+                if cond.is_none() {
+                    min_note_value(conn, table_prefix, account, fallback, chain_tip_height)
+                } else {
+                    Ok(cond)
+                }
+            }
+        }
+    }
+
+    // TODO: Simplify the query before executing it. Not worrying about this now because queries
+    // will be developer-configured, not end-user defined.
+    if let Some(min_value) =
+        min_note_value(conn, table_prefix, account, selector, chain_tip_height)?
+    {
+        let (note_count, total_value) = run_selection(min_value)?;
+
+        Ok(Some(PoolMeta::new(
+            note_count,
+            total_value.unwrap_or(NonNegativeAmount::ZERO),
+        )))
+    } else {
+        Ok(None)
+    }
 }
diff --git a/zcash_client_sqlite/src/wallet/init.rs b/zcash_client_sqlite/src/wallet/init.rs
index baed0a129..b5fbc9539 100644
--- a/zcash_client_sqlite/src/wallet/init.rs
+++ b/zcash_client_sqlite/src/wallet/init.rs
@@ -230,6 +230,9 @@ fn sqlite_client_error_to_wallet_migration_error(e: SqliteClientError) -> Wallet
         SqliteClientError::EphemeralAddressReuse(_, _) => {
             unreachable!("we don't do ephemeral address tracking")
         }
+        SqliteClientError::NoteFilterInvalid(_) => {
+            unreachable!("we don't do note selection in migrations")
+        }
     }
 }
 
diff --git a/zcash_client_sqlite/src/wallet/orchard.rs b/zcash_client_sqlite/src/wallet/orchard.rs
index 046b81f6b..1b0e0cacb 100644
--- a/zcash_client_sqlite/src/wallet/orchard.rs
+++ b/zcash_client_sqlite/src/wallet/orchard.rs
@@ -486,6 +486,11 @@ pub(crate) mod tests {
         testing::pool::scan_cached_blocks_detects_spends_out_of_order::<OrchardPoolTester>()
     }
 
+    #[test]
+    fn metadata_queries_exclude_unwanted_notes() {
+        testing::pool::metadata_queries_exclude_unwanted_notes::<OrchardPoolTester>()
+    }
+
     #[test]
     fn pool_crossing_required() {
         testing::pool::pool_crossing_required::<OrchardPoolTester, SaplingPoolTester>()
diff --git a/zcash_client_sqlite/src/wallet/sapling.rs b/zcash_client_sqlite/src/wallet/sapling.rs
index c9cf157a4..9652761b1 100644
--- a/zcash_client_sqlite/src/wallet/sapling.rs
+++ b/zcash_client_sqlite/src/wallet/sapling.rs
@@ -499,6 +499,11 @@ pub(crate) mod tests {
         testing::pool::scan_cached_blocks_detects_spends_out_of_order::<SaplingPoolTester>()
     }
 
+    #[test]
+    fn metadata_queries_exclude_unwanted_notes() {
+        testing::pool::metadata_queries_exclude_unwanted_notes::<SaplingPoolTester>()
+    }
+
     #[test]
     #[cfg(feature = "orchard")]
     fn pool_crossing_required() {
diff --git a/zcash_primitives/src/transaction/fees/zip317.rs b/zcash_primitives/src/transaction/fees/zip317.rs
index 63c8089d0..7ea41415c 100644
--- a/zcash_primitives/src/transaction/fees/zip317.rs
+++ b/zcash_primitives/src/transaction/fees/zip317.rs
@@ -78,8 +78,8 @@ impl FeeRule {
     ///
     /// Using this fee rule with
     /// ```compile_fail
-    /// marginal_fee < 5000 || grace_actions < 2 \
-    ///     || p2pkh_standard_input_size > P2PKH_STANDARD_INPUT_SIZE \
+    /// marginal_fee < 5000 || grace_actions < 2
+    ///     || p2pkh_standard_input_size > P2PKH_STANDARD_INPUT_SIZE
     ///     || p2pkh_standard_output_size > P2PKH_STANDARD_OUTPUT_SIZE
     /// ```
     /// violates ZIP 317, and might cause transactions built with it to fail.