deal with empty partition

datafusion/physical-plan/src/joins/dynamic_filters.rs

@@ -96,10 +96,14 @@ impl DynamicFilterInfo {
     ) -> Result<bool, DataFusionError> {
         let mut inner = self.inner.lock();
 
-        if inner.final_expr.is_some() {
+        // that indicates this partition of stream may contains no data, so we return
+        // or we have already have the final_expr
+        if inner.final_expr.is_some()
+            || (inner.processed_partitions.contains(&partition)
+                && records.num_rows() == 0)
+        {
             return Ok(true);
         }
-
         if !inner.processed_partitions.insert(partition) {
             return Ok(false);
         }

datafusion/physical-plan/src/joins/hash_join.rs

@@ -1335,7 +1335,7 @@ impl HashJoinStream {
         build_timer.done();
         // Merge the information to dynamic filters (if there is any) and check if it's finalized
         let filter_finalized = if let Some(filter_info) = &self.dynamic_filter_info {
-            filter_info.merge_batch_and_check_finalized(&left_data.batch)?
+            filter_info.merge_batch_and_check_finalized(left_data.batch())?
         } else {
             true // If there's no dynamic filter, we consider it as "finalized"
         };