refactor(engine): replace long argument lists with params structs

Author: farhan-syah

nodedb/src/data/executor/dispatch/columnar.rs

@@ -56,14 +56,16 @@ impl CoreLoop {
                 }
                 self.execute_columnar_insert(
                     task,
-                    collection,
-                    payload,
-                    format,
-                    *intent,
-                    on_conflict_updates,
-                    surrogates,
-                    schema_bytes,
-                    provenance.as_ref(),
+                    crate::data::executor::handlers::columnar_write::ColumnarInsertParams {
+                        collection,
+                        payload,
+                        format,
+                        intent: *intent,
+                        on_conflict_updates,
+                        surrogates,
+                        schema_bytes,
+                        provenance: provenance.as_ref(),
+                    },
                 )
             }
 

nodedb/src/data/executor/dispatch/crdt.rs

@@ -27,13 +27,15 @@ impl CoreLoop {
                 constraint_version_required,
             } => self.execute_crdt_apply(
                 task,
-                collection,
-                document_id,
-                delta,
-                *surrogate,
-                *peer_id,
-                provenance.as_ref(),
-                *constraint_version_required,
+                crate::data::executor::handlers::control::crdt::CrdtApplyParams {
+                    collection,
+                    document_id,
+                    delta,
+                    surrogate: *surrogate,
+                    peer_id: *peer_id,
+                    provenance: provenance.as_ref(),
+                    constraint_version_required: *constraint_version_required,
+                },
             ),
 
             CrdtOp::ImportSnapshot {

nodedb/src/data/executor/dispatch/document.rs

@@ -170,18 +170,20 @@ impl CoreLoop {
                 };
                 self.execute_document_scan(
                     task,
-                    tid,
-                    collection,
-                    *limit,
-                    *offset,
-                    sort_keys,
-                    filters,
-                    *distinct,
-                    projection,
-                    computed_columns,
-                    window_functions,
-                    mode,
-                    prefilter.as_ref(),
+                    crate::data::executor::handlers::document::read::scan::DocumentScanParams {
+                        tid,
+                        collection,
+                        limit: *limit,
+                        offset: *offset,
+                        sort_keys,
+                        filters,
+                        distinct: *distinct,
+                        projection,
+                        computed_columns_bytes: computed_columns,
+                        window_functions_bytes: window_functions,
+                        mode,
+                        prefilter: prefilter.as_ref(),
+                    },
                 )
             }
 
@@ -318,13 +320,15 @@ impl CoreLoop {
                 bitemporal,
             } => self.execute_register_document_collection(
                 task,
-                tid,
-                collection,
-                indexes,
-                *crdt_enabled,
-                storage_mode,
-                enforcement,
-                *bitemporal,
+                super::super::handlers::document::write::RegisterDocumentCollectionParams {
+                    tid,
+                    collection,
+                    indexes,
+                    crdt_enabled: *crdt_enabled,
+                    storage_mode,
+                    enforcement,
+                    bitemporal: *bitemporal,
+                },
             ),
 
             DocumentOp::IndexLookup {
@@ -389,13 +393,15 @@ impl CoreLoop {
                 predicate,
             } => self.execute_backfill_index(
                 task,
-                tid,
-                collection,
-                path,
-                *is_array,
-                *unique,
-                *case_insensitive,
-                predicate.as_deref(),
+                super::super::handlers::document::write::BackfillIndexParams {
+                    tid,
+                    collection,
+                    path,
+                    is_array: *is_array,
+                    unique: *unique,
+                    case_insensitive: *case_insensitive,
+                    predicate: predicate.as_deref(),
+                },
             ),
 
             DocumentOp::MaterializeScan {

nodedb/src/data/executor/dispatch/meta.rs

@@ -219,11 +219,13 @@ impl CoreLoop {
                 new_collection,
             } => self.execute_rename_collection(
                 task,
-                *tenant_id,
-                *old_database_id,
-                *new_database_id,
-                old_collection,
-                new_collection,
+                crate::data::executor::handlers::control::move_tenant::RenameCollectionParams {
+                    tenant_id: *tenant_id,
+                    old_database_id: *old_database_id,
+                    new_database_id: *new_database_id,
+                    old_collection,
+                    new_collection,
+                },
             ),
 
             MetaOp::StageWrite { plan } => self.execute_stage_write(task, tid, plan),

nodedb/src/data/executor/dispatch/query.rs

@@ -68,7 +68,16 @@ impl CoreLoop {
                 distinct,
                 ..
             } => self.execute_provider_scan(
-                task, rows, filters, projection, sort_keys, *limit, *offset, *distinct,
+                task,
+                crate::data::executor::handlers::provider_scan::ProviderScanParams {
+                    rows_bytes: rows,
+                    filters_bytes: filters,
+                    projection,
+                    sort_keys,
+                    limit: *limit,
+                    offset: *offset,
+                    distinct: *distinct,
+                },
             ),
 
             QueryOp::HashJoin {
@@ -186,14 +195,16 @@ impl CoreLoop {
                 limit,
             } => self.execute_recursive_scan(
                 task,
-                tid,
-                collection,
-                base_filters,
-                recursive_filters,
-                join_link.as_ref(),
-                *max_iterations,
-                *distinct,
-                *limit,
+                crate::data::executor::handlers::recursive::RecursiveScanParams {
+                    tid,
+                    collection,
+                    base_filters,
+                    recursive_filters,
+                    join_link: join_link.as_ref(),
+                    max_iterations: *max_iterations,
+                    distinct: *distinct,
+                    limit: *limit,
+                },
             ),
 
             QueryOp::RecursiveValue {
@@ -206,13 +217,15 @@ impl CoreLoop {
                 distinct,
             } => self.execute_recursive_value(
                 task,
-                cte_name,
-                columns,
-                init_exprs,
-                step_exprs,
-                condition.as_deref(),
-                *max_depth,
-                *distinct,
+                crate::data::executor::handlers::recursive_value::RecursiveValueParams {
+                    cte_name,
+                    columns,
+                    init_exprs,
+                    step_exprs,
+                    condition: condition.as_deref(),
+                    max_depth: *max_depth,
+                    distinct: *distinct,
+                },
             ),
 
             QueryOp::FacetCounts {

nodedb/src/data/executor/handlers/columnar_write/insert.rs

@@ -20,6 +20,18 @@ use nodedb_physical::physical_plan::document::UpdateValue;
 
 use super::schema::{ndb_field_to_value, row_values_to_object};
 
+/// Parameters for [`CoreLoop::execute_columnar_insert`].
+pub(in crate::data::executor) struct ColumnarInsertParams<'a> {
+    pub collection: &'a str,
+    pub payload: &'a [u8],
+    pub format: &'a str,
+    pub intent: ColumnarInsertIntent,
+    pub on_conflict_updates: &'a [(String, UpdateValue)],
+    pub surrogates: &'a [Surrogate],
+    pub schema_bytes: &'a [u8],
+    pub provenance: Option<&'a SyncProvenance>,
+}
+
 impl CoreLoop {
     /// Execute a columnar insert: write rows from MessagePack payload to
     /// `MutationEngine`, applying intent-specific semantics on duplicate
@@ -34,19 +46,21 @@ impl CoreLoop {
     /// `SyncAckResult{Applied}`.
     ///
     /// When `provenance` is `None` (SQL path), behave as before.
-    #[allow(clippy::too_many_arguments)]
     pub(in crate::data::executor) fn execute_columnar_insert(
         &mut self,
         task: &ExecutionTask,
-        collection: &str,
-        payload: &[u8],
-        _format: &str,
-        intent: ColumnarInsertIntent,
-        on_conflict_updates: &[(String, UpdateValue)],
-        surrogates: &[Surrogate],
-        schema_bytes: &[u8],
-        provenance: Option<&SyncProvenance>,
+        params: ColumnarInsertParams<'_>,
     ) -> Response {
+        let ColumnarInsertParams {
+            collection,
+            payload,
+            format: _format,
+            intent,
+            on_conflict_updates,
+            surrogates,
+            schema_bytes,
+            provenance,
+        } = params;
         // ── Sync idempotency gate (Data-Plane side) ──────────────────────────
         if let Some(prov) = provenance {
             let admit = self.sync_admit(prov);

nodedb/src/data/executor/handlers/columnar_write/mod.rs

@@ -11,6 +11,7 @@ pub mod read_prior;
 pub mod schema;
 pub mod spatial;
 
+pub(in crate::data::executor) use insert::ColumnarInsertParams;
 pub(in crate::data::executor) use schema::ndb_field_to_value;
 // `ensure_columnar_engine_schema` is an inherent `CoreLoop` method (defined
 // in `schema.rs`), called via `self.` — no re-export needed.

nodedb/src/data/executor/handlers/control/crdt.rs

@@ -16,6 +16,17 @@ use crate::engine::crdt::tenant_state::ValidatedApplyOutcome;
 use crate::data::executor::core_loop::CoreLoop;
 use crate::data::executor::task::ExecutionTask;
 
+/// Parameters for [`CoreLoop::execute_crdt_apply`].
+pub(in crate::data::executor) struct CrdtApplyParams<'a> {
+    pub collection: &'a str,
+    pub document_id: &'a str,
+    pub delta: &'a [u8],
+    pub surrogate: Surrogate,
+    pub peer_id: u64,
+    pub provenance: Option<&'a SyncProvenance>,
+    pub constraint_version_required: u64,
+}
+
 impl CoreLoop {
     pub(in crate::data::executor) fn execute_crdt_read(
         &mut self,
@@ -207,18 +218,20 @@ impl CoreLoop {
         }
     }
 
-    #[allow(clippy::too_many_arguments)]
     pub(in crate::data::executor) fn execute_crdt_apply(
         &mut self,
         task: &ExecutionTask,
-        collection: &str,
-        document_id: &str,
-        delta: &[u8],
-        surrogate: Surrogate,
-        peer_id: u64,
-        provenance: Option<&SyncProvenance>,
-        constraint_version_required: u64,
+        params: CrdtApplyParams<'_>,
     ) -> Response {
+        let CrdtApplyParams {
+            collection,
+            document_id,
+            delta,
+            surrogate,
+            peer_id,
+            provenance,
+            constraint_version_required,
+        } = params;
         let tenant_id = task.request.tenant_id;
 
         let Some(prov) = provenance else {

nodedb/src/data/executor/handlers/control/move_tenant.rs

@@ -11,20 +11,31 @@ use crate::bridge::envelope::{ErrorCode, Response};
 use crate::data::executor::core_loop::CoreLoop;
 use crate::data::executor::task::ExecutionTask;
 
+/// Parameters for [`CoreLoop::execute_rename_collection`].
+pub(in crate::data::executor) struct RenameCollectionParams<'a> {
+    pub tenant_id: u64,
+    pub old_database_id: u64,
+    pub new_database_id: u64,
+    pub old_collection: &'a str,
+    pub new_collection: &'a str,
+}
+
 impl CoreLoop {
     /// Handle `MetaOp::RenameCollection`: re-key all documents and secondary
     /// indexes from `old_collection` to `new_collection` for `tenant_id` in
     /// every engine that uses db-qualified collection names for keying.
-    #[allow(clippy::too_many_arguments)]
     pub(in crate::data::executor) fn execute_rename_collection(
         &mut self,
         task: &ExecutionTask,
-        tenant_id: u64,
-        old_database_id: u64,
-        new_database_id: u64,
-        old_collection: &str,
-        new_collection: &str,
+        params: RenameCollectionParams<'_>,
     ) -> Response {
+        let RenameCollectionParams {
+            tenant_id,
+            old_database_id,
+            new_database_id,
+            old_collection,
+            new_collection,
+        } = params;
         // Sparse engine (document schemaless + document strict).
         if let Err(e) = self.sparse.rename_collection(
             old_database_id,