feat: wire RDF bloom typed payloads

src/datanode/src/region_server/registrations.rs

@@ -161,22 +161,20 @@ fn record_unregister_outcome(outcome: RemoteDynFilterUpdateOutcome) {
 
 #[derive(Debug, Clone)]
 struct PendingDynFilterUpdate {
-    payload: Vec<u8>,
+    payload: DynFilterPayload,
     generation: u64,
     is_complete: bool,
 }
 
 impl PendingDynFilterUpdate {
     fn from_initial_reg(reg: &InitialDynFilterReg) -> Option<Self> {
         let snapshot = reg.initial_snapshot.as_ref()?;
-        match &snapshot.payload {
-            DynFilterPayload::Datafusion(payload) => Some(Self {
-                payload: payload.clone(),
-                generation: snapshot.generation,
-                is_complete: snapshot.is_complete,
-            }),
-            _ => None,
-        }
+        // Accept any payload variant for pending; FE may send non-Datafusion in the future.
+        Some(Self {
+            payload: snapshot.payload.clone(),
+            generation: snapshot.generation,
+            is_complete: snapshot.is_complete,
+        })
     }
 }
 
@@ -190,14 +188,20 @@ struct RemoteDynFilterEpochState {
 struct RemoteDynFilterState {
     filter: Arc<DynamicFilterPhysicalExpr>,
     input_schema: SchemaRef,
+    registered_children: Vec<Arc<dyn PhysicalExpr>>,
     epoch: Mutex<RemoteDynFilterEpochState>,
 }
 
 impl RemoteDynFilterState {
-    fn new(filter: Arc<DynamicFilterPhysicalExpr>, input_schema: SchemaRef) -> Self {
+    fn new(
+        filter: Arc<DynamicFilterPhysicalExpr>,
+        input_schema: SchemaRef,
+        registered_children: Vec<Arc<dyn PhysicalExpr>>,
+    ) -> Self {
         Self {
             filter,
             input_schema,
+            registered_children,
             epoch: Mutex::new(RemoteDynFilterEpochState {
                 generation: None,
                 is_complete: false,
@@ -211,7 +215,7 @@ impl RemoteDynFilterState {
 
     fn apply_update(
         &self,
-        payload: &[u8],
+        payload: &DynFilterPayload,
         generation: u64,
         is_complete: bool,
     ) -> RemoteDynFilterUpdateOutcome {
@@ -239,7 +243,12 @@ impl RemoteDynFilterState {
             return RemoteDynFilterUpdateOutcome::AlreadyComplete;
         }
 
-        let expr = match decode_update_payload(payload, self.input_schema.as_ref()) {
+        let expr = match payload.decode_expr_with_registered_children(
+            remote_dyn_filter_task_context(),
+            self.input_schema.as_ref(),
+            &self.registered_children,
+            REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
+        ) {
             Ok(expr) => expr,
             Err(error) => {
                 warn!(error; "Failed to decode remote dynamic filter update payload");
@@ -324,7 +333,7 @@ impl RegisteredDynFilter {
 
     fn buffer_update(
         &mut self,
-        payload: &[u8],
+        payload: &DynFilterPayload,
         generation: u64,
         is_complete: bool,
     ) -> RemoteDynFilterUpdateOutcome {
@@ -348,7 +357,7 @@ impl RegisteredDynFilter {
         }
 
         self.pending_update = Some(PendingDynFilterUpdate {
-            payload: payload.to_vec(),
+            payload: payload.clone(),
             generation,
             is_complete,
         });
@@ -357,7 +366,7 @@ impl RegisteredDynFilter {
 
     fn apply_or_buffer_update(
         &mut self,
-        payload: &[u8],
+        payload: &DynFilterPayload,
         generation: u64,
         is_complete: bool,
     ) -> RemoteDynFilterUpdateOutcome {
@@ -399,6 +408,7 @@ impl RegisteredDynFilter {
                 let runtime = Arc::new(RemoteDynFilterState::new(
                     filter,
                     Arc::new(input_schema.clone()),
+                    children.clone(),
                 ));
                 if let Some(pending) = self.pending_update.take() {
                     let outcome = runtime.apply_update(
@@ -537,7 +547,7 @@ pub(super) fn apply_remote_dyn_filter_update(
     regs_by_query: &RemoteDynFilterRegistry,
     query_id: &QueryId,
     filter_id: &RemoteDynFilterId,
-    payload: &[u8],
+    payload: &DynFilterPayload,
     generation: u64,
     is_complete: bool,
 ) -> RemoteDynFilterUpdateOutcome {
@@ -546,15 +556,15 @@ pub(super) fn apply_remote_dyn_filter_update(
             "Ignored oversized remote dynamic filter update, query_id: {}, filter_id: {}, payload_size: {}, max_payload_size: {}",
             query_id,
             filter_id,
-            payload.len(),
+            payload.encoded_payload_bytes(),
             REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES
         );
         let outcome = RemoteDynFilterUpdateOutcome::PayloadTooLarge;
         record_update_outcome(outcome);
         return outcome;
     }
 
-    REMOTE_DYN_FILTER_PAYLOAD_BYTES.observe(payload.len() as f64);
+    REMOTE_DYN_FILTER_PAYLOAD_BYTES.observe(payload.encoded_payload_bytes() as f64);
 
     let Some(query_regs) = regs_by_query.get_query(query_id) else {
         warn!(
@@ -665,15 +675,8 @@ pub(super) fn remove_initial_dyn_filter_regs(
     }
 }
 
-fn decode_update_payload(
-    payload: &[u8],
-    input_schema: &Schema,
-) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
-    DynFilterPayload::Datafusion(payload.to_vec()).decode_datafusion_expr(
-        remote_dyn_filter_task_context(),
-        input_schema,
-        REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
-    )
+fn validate_update_payload_size(payload: &DynFilterPayload) -> bool {
+    payload.encoded_payload_bytes() <= REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES
 }
 
 fn remote_dyn_filter_task_context() -> &'static TaskContext {
@@ -689,10 +692,6 @@ fn remote_dyn_filter_task_context() -> &'static TaskContext {
     })
 }
 
-fn validate_update_payload_size(payload: &[u8]) -> bool {
-    payload.len() <= REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;