fix(delete): preserve parent type for non-empty subtree deletes

grovedb-version/src/version/v3.rs

@@ -117,7 +117,7 @@ pub const GROVE_V3: GroveVersion = GroveVersion {
                 delete_if_empty_tree: 0,
                 delete_if_empty_tree_with_sectional_storage_function: 0,
                 delete_operation_for_delete_internal: 0,
-                delete_internal_on_transaction: 0,
+                delete_internal_on_transaction: 1,
                 delete_internal_without_transaction: 0,
                 average_case_delete_operation_for_delete: 0,
                 worst_case_delete_operation_for_delete: 0,

grovedb/src/operations/delete/delete_internal_on_transaction/mod.rs

@@ -0,0 +1,79 @@
+//! Versioned dispatch for [`GroveDb::delete_internal_on_transaction`].
+//!
+//! Each `v*.rs` defines one `delete_internal_on_transaction_v*` method. The
+//! dispatcher below selects the implementation by
+//! `GroveVersion::grovedb_versions.operations.delete.delete_internal_on_transaction`.
+//!
+//! `v0` preserves the historical (grove v1/v2) behavior: when deleting a
+//! non-empty tree it reopened the parent layer using the *child* tree type,
+//! which propagated parent aggregate (count/sum) metadata incorrectly. `v1`
+//! (grove v3+) reuses the already-open parent merk so propagation uses the
+//! parent tree type. See issue #686.
+
+mod v0;
+mod v1;
+
+use grovedb_costs::{
+    storage_cost::removal::StorageRemovedBytes, CostResult, CostsExt, OperationCost,
+};
+use grovedb_merk::Error as MerkError;
+use grovedb_path::SubtreePath;
+use grovedb_storage::StorageBatch;
+use grovedb_version::{error::GroveVersionError, version::GroveVersion};
+
+use super::DeleteOptions;
+use crate::{Error, GroveDb, Transaction};
+
+impl GroveDb {
+    pub(super) fn delete_internal_on_transaction<B: AsRef<[u8]>>(
+        &self,
+        path: SubtreePath<B>,
+        key: &[u8],
+        options: &DeleteOptions,
+        transaction: &Transaction,
+        sectioned_removal: &mut impl FnMut(
+            &Vec<u8>,
+            u32,
+            u32,
+        ) -> Result<
+            (StorageRemovedBytes, StorageRemovedBytes),
+            MerkError,
+        >,
+        batch: &StorageBatch,
+        grove_version: &GroveVersion,
+    ) -> CostResult<bool, Error> {
+        match grove_version
+            .grovedb_versions
+            .operations
+            .delete
+            .delete_internal_on_transaction
+        {
+            0 => self.delete_internal_on_transaction_v0(
+                path,
+                key,
+                options,
+                transaction,
+                sectioned_removal,
+                batch,
+                grove_version,
+            ),
+            1 => self.delete_internal_on_transaction_v1(
+                path,
+                key,
+                options,
+                transaction,
+                sectioned_removal,
+                batch,
+                grove_version,
+            ),
+            version => Err(Error::VersionError(
+                GroveVersionError::UnknownVersionMismatch {
+                    method: "delete_internal_on_transaction".to_string(),
+                    known_versions: vec![0, 1],
+                    received: version,
+                },
+            ))
+            .wrap_with_cost(OperationCost::default()),
+        }
+    }
+}

grovedb/src/operations/delete/delete_internal_on_transaction/v0.rs

@@ -0,0 +1,262 @@
+//! `delete_internal_on_transaction` v0 — grove v1/v2 dispatch path
+//! (historical baseline).
+//!
+//! When deleting a non-empty tree element, this reopens the parent layer using
+//! the *child* tree type (`tree_type`) before deleting. This propagates parent
+//! aggregate/root metadata incorrectly for aggregate parents (CountTree,
+//! SumTree, CountSumTree, ...). It is preserved verbatim for replay
+//! compatibility; the corrected behavior lives in
+//! [`super::v1`](super::v1) (grove v3+). See issue #686.
+
+use std::collections::HashMap;
+
+use grovedb_costs::{
+    cost_return_on_error, cost_return_on_error_into, storage_cost::removal::StorageRemovedBytes,
+    CostResult, CostsExt, OperationCost,
+};
+use grovedb_merk::{
+    element::{
+        costs::ElementCostExtensions, delete::ElementDeleteFromStorageExtensions,
+        tree_type::ElementTreeTypeExtensions,
+    },
+    Error as MerkError, Merk,
+};
+use grovedb_path::SubtreePath;
+use grovedb_storage::{rocksdb_storage::PrefixedRocksDbTransactionContext, Storage, StorageBatch};
+use grovedb_version::version::GroveVersion;
+
+use super::super::DeleteOptions;
+use crate::{Element, Error, GroveDb, Transaction};
+
+impl GroveDb {
+    pub(super) fn delete_internal_on_transaction_v0<B: AsRef<[u8]>>(
+        &self,
+        path: SubtreePath<B>,
+        key: &[u8],
+        options: &DeleteOptions,
+        transaction: &Transaction,
+        sectioned_removal: &mut impl FnMut(
+            &Vec<u8>,
+            u32,
+            u32,
+        ) -> Result<
+            (StorageRemovedBytes, StorageRemovedBytes),
+            MerkError,
+        >,
+        batch: &StorageBatch,
+        grove_version: &GroveVersion,
+    ) -> CostResult<bool, Error> {
+        let mut cost = OperationCost::default();
+
+        let element = cost_return_on_error!(
+            &mut cost,
+            self.get_raw(path.clone(), key.as_ref(), Some(transaction), grove_version)
+        );
+        let mut subtree_to_delete_from = cost_return_on_error!(
+            &mut cost,
+            self.open_transactional_merk_at_path(
+                path.clone(),
+                transaction,
+                Some(batch),
+                grove_version
+            )
+        );
+        let parent_tree_type = subtree_to_delete_from.tree_type;
+        if let Some(tree_type) = element.tree_type() {
+            let subtree_merk_path = path.derive_owned_with_child(key);
+            let subtree_merk_path_ref = SubtreePath::from(&subtree_merk_path);
+
+            // Tree types that store data in the data namespace as non-Merk
+            // entries (CommitmentTree, MmrTree, BulkAppendTree, DenseTree)
+            // have an always-empty Merk but may have data.  We cannot iterate
+            // their storage with find_subtrees because the entries are not
+            // valid Element serializations.
+            let non_merk_data = element.uses_non_merk_data_storage();
+
+            let subtree_of_tree_we_are_deleting = cost_return_on_error!(
+                &mut cost,
+                self.open_transactional_merk_at_path(
+                    subtree_merk_path_ref.clone(),
+                    transaction,
+                    Some(batch),
+                    grove_version,
+                )
+            );
+
+            // For non-Merk data trees the raw_iter check would see non-Merk
+            // keys and wrongly report the tree as non-empty.  Use the
+            // element's own count instead.
+            let is_empty = if non_merk_data {
+                element.non_merk_entry_count().unwrap_or(0) == 0
+            } else {
+                subtree_of_tree_we_are_deleting
+                    .is_empty_tree()
+                    .unwrap_add_cost(&mut cost)
+            };
+
+            if !options.allow_deleting_non_empty_trees && !is_empty {
+                return if options.deleting_non_empty_trees_returns_error {
+                    Err(Error::DeletingNonEmptyTree(
+                        "trying to do a delete operation for a non empty tree, but options not \
+                         allowing this",
+                    ))
+                    .wrap_with_cost(cost)
+                } else {
+                    Ok(false).wrap_with_cost(cost)
+                };
+            }
+
+            if !is_empty {
+                if non_merk_data {
+                    // Non-Merk data trees: clear the subtree storage directly.
+                    // These trees never contain child subtrees so we only need
+                    // to clear the one storage context.
+                    let mut storage = self
+                        .db
+                        .get_transactional_storage_context(
+                            subtree_merk_path_ref.clone(),
+                            Some(batch),
+                            transaction,
+                        )
+                        .unwrap_add_cost(&mut cost);
+                    cost_return_on_error!(
+                        &mut cost,
+                        storage.clear().map_err(|e| {
+                            Error::CorruptedData(format!(
+                                "unable to cleanup non-merk tree data from storage: {e}",
+                            ))
+                        })
+                    );
+                } else {
+                    let subtrees_paths = cost_return_on_error!(
+                        &mut cost,
+                        self.find_subtrees(
+                            &subtree_merk_path_ref,
+                            Some(transaction),
+                            grove_version
+                        )
+                    );
+                    for subtree_path in subtrees_paths {
+                        let p: SubtreePath<_> = subtree_path.as_slice().into();
+                        let mut storage = self
+                            .db
+                            .get_transactional_storage_context(p, Some(batch), transaction)
+                            .unwrap_add_cost(&mut cost);
+
+                        cost_return_on_error!(
+                            &mut cost,
+                            storage.clear().map_err(|e| {
+                                Error::CorruptedData(format!(
+                                    "unable to cleanup tree from storage: {e}",
+                                ))
+                            })
+                        );
+                    }
+                }
+                // Legacy behavior reopened the parent layer using the
+                // child tree type before deleting the tree element.
+                let storage = self
+                    .db
+                    .get_transactional_storage_context(path.clone(), Some(batch), transaction)
+                    .unwrap_add_cost(&mut cost);
+
+                let mut merk_to_delete_tree_from = cost_return_on_error!(
+                    &mut cost,
+                    Merk::open_layered_with_root_key(
+                        storage,
+                        subtree_to_delete_from.root_key(),
+                        tree_type,
+                        Some(&Element::value_defined_cost_for_serialized_value),
+                        grove_version,
+                    )
+                    .map_err(super::super::cannot_open_subtree_with_root_key_error)
+                );
+                // We are deleting a tree, a tree uses 3 bytes
+                cost_return_on_error_into!(
+                    &mut cost,
+                    Element::delete_with_sectioned_removal_bytes(
+                        &mut merk_to_delete_tree_from,
+                        key,
+                        Some(options.as_merk_options()),
+                        true,
+                        parent_tree_type,
+                        sectioned_removal,
+                        grove_version,
+                    )
+                );
+                let mut merk_cache: HashMap<
+                    SubtreePath<B>,
+                    Merk<PrefixedRocksDbTransactionContext>,
+                > = HashMap::default();
+                merk_cache.insert(path.clone(), merk_to_delete_tree_from);
+                cost_return_on_error!(
+                    &mut cost,
+                    self.propagate_changes_with_batch_transaction(
+                        batch,
+                        merk_cache,
+                        &path,
+                        transaction,
+                        grove_version,
+                    )
+                );
+            } else {
+                // We are deleting a tree, a tree uses 3 bytes
+                cost_return_on_error_into!(
+                    &mut cost,
+                    Element::delete_with_sectioned_removal_bytes(
+                        &mut subtree_to_delete_from,
+                        key,
+                        Some(options.as_merk_options()),
+                        true,
+                        parent_tree_type,
+                        sectioned_removal,
+                        grove_version,
+                    )
+                );
+                let mut merk_cache: HashMap<
+                    SubtreePath<B>,
+                    Merk<PrefixedRocksDbTransactionContext>,
+                > = HashMap::default();
+                merk_cache.insert(path.clone(), subtree_to_delete_from);
+                cost_return_on_error!(
+                    &mut cost,
+                    self.propagate_changes_with_transaction(
+                        merk_cache,
+                        path,
+                        transaction,
+                        batch,
+                        grove_version
+                    )
+                );
+            }
+        } else {
+            cost_return_on_error_into!(
+                &mut cost,
+                Element::delete_with_sectioned_removal_bytes(
+                    &mut subtree_to_delete_from,
+                    key,
+                    Some(options.as_merk_options()),
+                    false,
+                    parent_tree_type,
+                    sectioned_removal,
+                    grove_version,
+                )
+            );
+            let mut merk_cache: HashMap<SubtreePath<B>, Merk<PrefixedRocksDbTransactionContext>> =
+                HashMap::default();
+            merk_cache.insert(path.clone(), subtree_to_delete_from);
+            cost_return_on_error!(
+                &mut cost,
+                self.propagate_changes_with_transaction(
+                    merk_cache,
+                    path,
+                    transaction,
+                    batch,
+                    grove_version
+                )
+            );
+        }
+
+        Ok(true).wrap_with_cost(cost)
+    }
+}