refactor(rust): streaming improvements (#5541)

Author: ritchie46

polars/polars-core/src/datatypes/any_value.rs

@@ -430,52 +430,62 @@ impl<'a> Hash for AnyValue<'a> {
 impl<'a> Eq for AnyValue<'a> {}
 
 impl From<f64> for AnyValue<'_> {
+    #[inline]
     fn from(a: f64) -> Self {
         AnyValue::Float64(a)
     }
 }
 impl From<f32> for AnyValue<'_> {
+    #[inline]
     fn from(a: f32) -> Self {
         AnyValue::Float32(a)
     }
 }
 impl From<u32> for AnyValue<'_> {
+    #[inline]
     fn from(a: u32) -> Self {
         AnyValue::UInt32(a)
     }
 }
 impl From<u64> for AnyValue<'_> {
+    #[inline]
     fn from(a: u64) -> Self {
         AnyValue::UInt64(a)
     }
 }
 impl From<i64> for AnyValue<'_> {
+    #[inline]
     fn from(a: i64) -> Self {
         AnyValue::Int64(a)
     }
 }
 impl From<i32> for AnyValue<'_> {
+    #[inline]
     fn from(a: i32) -> Self {
         AnyValue::Int32(a)
     }
 }
 impl From<i16> for AnyValue<'_> {
+    #[inline]
     fn from(a: i16) -> Self {
         AnyValue::Int16(a)
     }
 }
 impl From<u16> for AnyValue<'_> {
+    #[inline]
     fn from(a: u16) -> Self {
         AnyValue::UInt16(a)
     }
 }
 
 impl From<i8> for AnyValue<'_> {
+    #[inline]
     fn from(a: i8) -> Self {
         AnyValue::Int8(a)
     }
 }
 impl From<u8> for AnyValue<'_> {
+    #[inline]
     fn from(a: u8) -> Self {
         AnyValue::UInt8(a)
     }
@@ -485,6 +495,7 @@ impl<'a, T> From<Option<T>> for AnyValue<'a>
 where
     T: Into<AnyValue<'a>>,
 {
+    #[inline]
     fn from(a: Option<T>) -> Self {
         match a {
             None => AnyValue::Null,

polars/polars-core/src/frame/hash_join/mod.rs

@@ -111,7 +111,7 @@ pub(crate) fn check_categorical_src(l: &DataType, r: &DataType) -> PolarsResult<
         (DataType::Categorical(Some(l)), DataType::Categorical(Some(r))) => {
             if !l.same_src(r) {
                 return Err(PolarsError::ComputeError("Joins/or comparisons on categorical dtypes can only happen if they are created under the same global string cache.\
-                 Hint: set a global StringCache".into()));
+                  Hint: set a global StringCache".into()));
             }
             Ok(())
         }

polars/polars-core/src/vector_hasher.rs

@@ -125,10 +125,14 @@ impl VecHash for Utf8Chunked {
         buf.reserve(self.len());
         let null_h = get_null_hash_value(random_state.clone());
         self.downcast_iter().for_each(|arr| {
-            buf.extend(arr.into_iter().map(|opt_v| match opt_v {
-                Some(v) => random_state.hash_single(v),
-                None => null_h,
-            }))
+            if arr.null_count() == 0 {
+                buf.extend(arr.values_iter().map(|v| random_state.hash_single(v)))
+            } else {
+                buf.extend(arr.into_iter().map(|opt_v| match opt_v {
+                    Some(v) => random_state.hash_single(v),
+                    None => null_h,
+                }))
+            }
         });
     }
 

polars/polars-lazy/polars-pipe/src/executors/sinks/groupby/aggregates/convert.rs

@@ -89,7 +89,7 @@ where
 {
     match expr_arena.get(node) {
         AExpr::Alias(input, _) => convert_to_hash_agg(*input, expr_arena, schema, to_physical),
-        AExpr::Count => (
+        AExpr::Count | AExpr::Agg(AAggExpr::Count(_)) => (
             Arc::new(Count {}),
             AggregateFunction::Count(CountAgg::new()),
         ),
@@ -143,7 +143,7 @@ where
                 let dtype = phys_expr.field(schema).unwrap().dtype;
                 (phys_expr, AggregateFunction::Last(LastAgg::new(dtype)))
             }
-            _ => todo!(),
+            agg => panic!("{:?} not yet implemented.", agg),
         },
         _ => todo!(),
     }