fixed filtering of nulls and mixed types including AnyValues

Author: albertlockett

rust/otap-dataflow/crates/query-engine/src/pipeline/assign.rs

@@ -61,8 +61,8 @@ use crate::pipeline::expr::{
 };
 use crate::pipeline::planner::{AttributesIdentifier, ColumnAccessor};
 use crate::pipeline::project::anyval::{
-    fill_null_type_as_empty, is_any_value_data_type, project_any_value_columns,
-    wrap_as_any_value_struct,
+    attempt_coerce_value_column_from_any_value_struct_column, fill_null_type_as_empty,
+    is_any_value_data_type, wrap_as_any_value_struct,
 };
 use crate::pipeline::project::{ProjectedSchemaColumn, Projection};
 use crate::pipeline::state::ExecutionState;
@@ -1660,32 +1660,6 @@ fn coerce_to_any_value_struct_column(values: ArrayRef) -> Result<ArrayRef> {
     }
 }
 
-/// Attempt to coerce the values array (a struct array representing an AnyValue) into the concrete
-/// values type. If the passed array contains multiple types, the original array is returned
-/// because coercion was not successful.
-fn attempt_coerce_value_column_from_any_value_struct_column(values: &ArrayRef) -> Result<ArrayRef> {
-    // build a temporary record batch containing the column to maybe partition by type
-    let rb = RecordBatch::try_new(
-        Arc::new(Schema::new(vec![Field::new(
-            "",
-            values.data_type().clone(),
-            true,
-        )])),
-        vec![Arc::clone(values)],
-    )?;
-
-    // attempt to partition the AnyValue column by type
-    let partitions = project_any_value_columns(&rb, &[0])?;
-    let result = if partitions.len() == 1 {
-        partitions[0].batch.column(0)
-    } else {
-        // just return the original array
-        values
-    };
-
-    Ok(Arc::clone(result))
-}
-
 /// Inserts the column into the record batch if the column does not exist, otherwise replaces the
 /// existing column with the new one.
 ///