fix: propagate post_dedup field across all PhysicalTask construction sites

All locations that build a PhysicalTask now set post_dedup: false
explicitly to satisfy the new required field.

Also fixes the sync session to derive current_lsn from the tracked
server clock rather than returning a hardcoded zero, and updates
the columnar mutation comment to reflect the scan-based UPDATE path
that is actually implemented.

Author: farhan-syah

nodedb/src/control/planner/auto_tier.rs

@@ -225,6 +225,7 @@ fn build_scan_task(
             gap_fill: gap_fill.to_string(),
             rls_filters: Vec::new(),
         }),
+        post_dedup: false,
     }
 }
 

nodedb/src/control/planner/procedural/executor/transaction.rs

@@ -106,6 +106,7 @@ mod tests {
                 document_id: id.into(),
                 value: vec![],
             }),
+            post_dedup: false,
         }
     }
 

nodedb/src/control/server/native/dispatch/transaction.rs

@@ -87,6 +87,7 @@ pub(crate) async fn handle_commit(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResp
             tenant_id,
             vshard_id,
             plan: PhysicalPlan::Meta(MetaOp::TransactionBatch { plans }),
+            post_dedup: false,
         };
         if let Err(e) = dispatch_utils::dispatch_to_data_plane(
             ctx.state,

nodedb/src/control/server/pgwire/handler/dispatch.rs

@@ -67,6 +67,7 @@ impl NodeDbPgHandler {
                 limit,
                 ref post_group_by,
                 ref post_aggregates,
+                ..
             },
         ) = task.plan
         {

nodedb/src/control/server/pgwire/handler/facet.rs

@@ -46,6 +46,7 @@ pub(super) async fn execute_facet_counts_sql(
             fields: parsed.fields,
             limit_per_facet: parsed.limit_per_facet,
         }),
+        post_dedup: false,
     };
 
     let resp = handler.dispatch_task(task).await.map_err(|e| {
@@ -98,6 +99,7 @@ pub(super) async fn execute_search_with_facets_sql(
             fields: parsed.facets,
             limit_per_facet: 0, // All values.
         }),
+        post_dedup: false,
     };
 
     let facet_resp = handler.dispatch_task(facet_task).await.map_err(|e| {

nodedb/src/control/server/pgwire/handler/prepared/plan_cache.rs

@@ -152,6 +152,7 @@ mod tests {
             tenant_id: TenantId::new(1),
             vshard_id: VShardId::new(0),
             plan: PhysicalPlan::Meta(MetaOp::Checkpoint),
+            post_dedup: false,
         }]
     }
 

nodedb/src/control/server/pgwire/handler/transaction_cmds.rs

@@ -120,6 +120,7 @@ impl NodeDbPgHandler {
                 plan: crate::bridge::envelope::PhysicalPlan::Meta(
                     crate::bridge::physical_plan::MetaOp::TransactionBatch { plans },
                 ),
+                post_dedup: false,
             };
             if let Err(e) = self.dispatch_task_no_wal(batch_task).await {
                 tracing::warn!(error = %e, "transaction batch dispatch failed");

nodedb/src/control/server/sync/session.rs

@@ -560,7 +560,7 @@ impl SyncSession {
                 let msg: super::shape::handler::ShapeSubscribeMsg = frame.decode_body()?;
                 let registry = super::shape::registry::ShapeRegistry::new();
                 let tenant_id = self.tenant_id.map(|t| t.as_u32()).unwrap_or(0);
-                let current_lsn = 0u64; // TODO: get from WAL when wired
+                let current_lsn = self.server_clock.values().copied().max().unwrap_or(0);
                 let response = super::shape::handler::handle_subscribe(
                     &self.session_id,
                     tenant_id,

nodedb/src/data/executor/handlers/columnar_mutation.rs

@@ -37,8 +37,8 @@ impl CoreLoop {
             }
         };
 
-        // For now, columnar UPDATE requires PK-based access.
-        // TODO: scan-based UPDATE with filter predicates.
+        // Columnar UPDATE: scan memtable rows matching filter predicates,
+        // then apply updates via PK-based MutationEngine (delete + re-insert).
         let schema = engine.schema().clone();
         let pk_cols: Vec<usize> = schema
             .columns

nodedb/tests/cluster_procedures.rs

@@ -75,6 +75,7 @@ fn tx_ctx_commit_yields_independent_tasks() {
             document_id: "o-1".into(),
             value: b"{}".to_vec(),
         }),
+        post_dedup: false,
     });
     ctx.buffer_task(nodedb::control::planner::physical::PhysicalTask {
         tenant_id: TenantId::new(1),
@@ -83,6 +84,7 @@ fn tx_ctx_commit_yields_independent_tasks() {
             collection: "temp".into(),
             document_id: "t-1".into(),
         }),
+        post_dedup: false,
     });
 
     // COMMIT flushes both tasks.
@@ -119,6 +121,7 @@ fn procedure_can_target_multiple_vshards() {
             document_id: "d1".into(),
             value: vec![],
         }),
+        post_dedup: false,
     });
     // Task on vshard 1 (different shard)
     ctx.buffer_task(nodedb::control::planner::physical::PhysicalTask {
@@ -129,6 +132,7 @@ fn procedure_can_target_multiple_vshards() {
             document_id: "d2".into(),
             value: vec![],
         }),
+        post_dedup: false,
     });
 
     let tasks = ctx.take_buffered_tasks();