feat: add projection/filters to SqlPlan::Join and post_dedup to PhysicalTask

SqlPlan::Join now carries post-join projection and filter fields so the
planner can propagate SELECT column lists and WHERE predicates through
join plans rather than discarding them.

PhysicalTask gains a post_dedup flag used by the router to trigger
UNION DISTINCT row deduplication after collecting responses from all
sub-query tasks.

QueryOp::HashJoin picks up matching projection and post_filters fields
for the Data Plane executor.

Author: farhan-syah

nodedb-sql/src/planner/select.rs

@@ -126,6 +126,8 @@ fn plan_select(
                 join_type: sq.join_type,
                 condition: None,
                 limit: 10000,
+                projection: Vec::new(),
+                filters: Vec::new(),
             };
         }
         return Ok(plan);
@@ -159,6 +161,8 @@ fn plan_select(
             join_type: sq.join_type,
             condition: None,
             limit: 10000,
+            projection: Vec::new(),
+            filters: Vec::new(),
         };
     }
 

nodedb-sql/src/types.rs

@@ -77,6 +77,10 @@ pub enum SqlPlan {
         join_type: JoinType,
         condition: Option<SqlExpr>,
         limit: usize,
+        /// Post-join projection: column names to keep (empty = all columns).
+        projection: Vec<Projection>,
+        /// Post-join filters (from WHERE clause).
+        filters: Vec<Filter>,
     },
 
     // ── Aggregation ──

nodedb/src/bridge/physical_plan/query.rs

@@ -35,6 +35,10 @@ pub enum QueryOp {
         post_group_by: Vec<String>,
         /// Post-join aggregates: (op, field) pairs (empty = no aggregation).
         post_aggregates: Vec<(String, String)>,
+        /// Post-join projection: column names to keep (empty = all).
+        projection: Vec<String>,
+        /// Post-join WHERE filter predicates (MessagePack).
+        post_filters: Vec<u8>,
     },
 
     /// Broadcast join: small side serialized in the plan.

nodedb/src/control/planner/physical.rs

@@ -15,4 +15,8 @@ pub struct PhysicalTask {
 
     /// The physical operation to execute.
     pub plan: PhysicalPlan,
+
+    /// When true, results from this task batch must be deduplicated
+    /// (e.g. UNION DISTINCT across multiple sub-queries).
+    pub post_dedup: bool,
 }