vortex_array/scalar_fn/fns/fill_null/
mod.rs1mod kernel;
5
6pub use kernel::*;
7use vortex_error::VortexResult;
8use vortex_error::vortex_bail;
9use vortex_error::vortex_ensure;
10use vortex_error::vortex_err;
11use vortex_session::VortexSession;
12use vortex_session::registry::CachedId;
13
14use crate::AnyColumnar;
15use crate::ArrayRef;
16use crate::CanonicalView;
17use crate::ColumnarView;
18use crate::ExecutionCtx;
19use crate::IntoArray;
20use crate::arrays::Bool;
21use crate::arrays::Decimal;
22use crate::arrays::Primitive;
23use crate::arrays::ScalarFnArray;
24use crate::dtype::DType;
25use crate::expr::BoundExpression;
26use crate::scalar::Scalar;
27use crate::scalar_fn::Arity;
28use crate::scalar_fn::ChildName;
29use crate::scalar_fn::EmptyOptions;
30use crate::scalar_fn::ExecutionArgs;
31use crate::scalar_fn::ReduceNode;
32use crate::scalar_fn::ReduceNodeValidity;
33use crate::scalar_fn::ScalarFnId;
34use crate::scalar_fn::ScalarFnVTable;
35use crate::scalar_fn::ScalarFnVTableExt;
36use crate::scalar_fn::is_not_null_node;
37
38#[derive(Clone)]
40pub struct FillNull;
41
42impl FillNull {
43 pub fn try_new(input: ArrayRef, fill_value: ArrayRef) -> VortexResult<ScalarFnArray> {
49 ScalarFnArray::try_new(FillNull.bind(EmptyOptions), [input, fill_value])
50 }
51}
52
53impl ScalarFnVTable for FillNull {
54 type Options = EmptyOptions;
55
56 fn id(&self) -> ScalarFnId {
57 static ID: CachedId = CachedId::new("vortex.fill_null");
58 *ID
59 }
60
61 fn serialize(&self, _options: &Self::Options) -> VortexResult<Option<Vec<u8>>> {
62 Ok(Some(vec![]))
63 }
64
65 fn deserialize(
66 &self,
67 _metadata: &[u8],
68 _session: &VortexSession,
69 ) -> VortexResult<Self::Options> {
70 Ok(EmptyOptions)
71 }
72
73 fn arity(&self, _options: &Self::Options) -> Arity {
74 Arity::Exact(2)
75 }
76
77 fn child_name(&self, _options: &Self::Options, child_idx: usize) -> ChildName {
78 match child_idx {
79 0 => ChildName::from("input"),
80 1 => ChildName::from("fill_value"),
81 _ => unreachable!("Invalid child index {} for FillNull expression", child_idx),
82 }
83 }
84
85 fn return_dtype(&self, _options: &Self::Options, arg_dtypes: &[DType]) -> VortexResult<DType> {
86 vortex_ensure!(
87 arg_dtypes[0].eq_ignore_nullability(&arg_dtypes[1]),
88 "fill_null requires input and fill value to have the same base type, got {} and {}",
89 arg_dtypes[0],
90 arg_dtypes[1]
91 );
92 Ok(arg_dtypes[0]
94 .clone()
95 .with_nullability(arg_dtypes[1].nullability()))
96 }
97
98 fn execute(
99 &self,
100 _options: &Self::Options,
101 args: &dyn ExecutionArgs,
102 ctx: &mut ExecutionCtx,
103 ) -> VortexResult<ArrayRef> {
104 let input = args.get(0)?;
105 let fill_value = args.get(1)?;
106
107 let fill_scalar = fill_value
108 .as_constant()
109 .ok_or_else(|| vortex_err!("fill_null fill_value must be a constant/scalar"))?;
110
111 vortex_ensure!(
112 !fill_scalar.is_null(),
113 "fill_null requires a non-null fill value"
114 );
115
116 let Some(columnar) = input.as_opt::<AnyColumnar>() else {
117 let input = input.execute::<ArrayRef>(ctx)?;
118 return Ok(FillNull::try_new(input, fill_value)?.into_array());
119 };
120
121 match columnar {
122 ColumnarView::Canonical(canonical) => fill_null_canonical(canonical, &fill_scalar, ctx),
123 ColumnarView::Constant(constant) => fill_null_constant(constant, &fill_scalar),
124 }
125 }
126
127 fn simplify(
128 &self,
129 _options: &Self::Options,
130 expr: &BoundExpression,
131 ) -> VortexResult<Option<BoundExpression>> {
132 if !expr.child(0).dtype().is_nullable() {
133 return Ok(Some(expr.child(0).clone()));
134 }
135
136 Ok(None)
137 }
138
139 fn validity<T: ReduceNode>(
140 &self,
141 _options: &Self::Options,
142 node: &T,
143 ) -> VortexResult<ReduceNodeValidity<T>> {
144 Ok(ReduceNodeValidity::Reduced(is_not_null_node(
147 &node.child(1),
148 )?))
149 }
150
151 fn is_strict(&self, _options: &Self::Options) -> bool {
152 false
154 }
155
156 fn is_infallible(&self, _options: &Self::Options) -> bool {
157 true
158 }
159}
160
161fn fill_null_canonical(
165 canonical: CanonicalView<'_>,
166 fill_value: &Scalar,
167 ctx: &mut ExecutionCtx,
168) -> VortexResult<ArrayRef> {
169 let arr = canonical.to_array_ref();
170 if let Some(result) = short_circuit(&arr, fill_value)? {
171 return Ok(result);
172 }
173 match canonical {
174 CanonicalView::Bool(a) => <Bool as FillNullKernel>::fill_null(a, fill_value, ctx)?
175 .ok_or_else(|| vortex_err!("FillNullKernel for BoolArray returned None")),
176 CanonicalView::Primitive(a) => {
177 <Primitive as FillNullKernel>::fill_null(a, fill_value, ctx)?
178 .ok_or_else(|| vortex_err!("FillNullKernel for PrimitiveArray returned None"))
179 }
180 CanonicalView::Decimal(a) => <Decimal as FillNullKernel>::fill_null(a, fill_value, ctx)?
181 .ok_or_else(|| vortex_err!("FillNullKernel for DecimalArray returned None")),
182 other => vortex_bail!(
183 "No FillNullKernel for canonical array {}",
184 other.to_array_ref().encoding_id()
185 ),
186 }
187}
188
189#[cfg(test)]
190mod tests {
191 use vortex_buffer::buffer;
192 use vortex_error::VortexExpect;
193
194 use crate::IntoArray;
195 use crate::VortexSessionExecute;
196 use crate::array_session;
197 use crate::arrays::PrimitiveArray;
198 use crate::arrays::StructArray;
199 use crate::assert_arrays_eq;
200 use crate::dtype::DType;
201 use crate::dtype::Nullability;
202 use crate::dtype::PType;
203 use crate::expr::fill_null;
204 use crate::expr::get_item;
205 use crate::expr::lit;
206 use crate::expr::root;
207
208 #[test]
209 fn dtype() {
210 let dtype = DType::Primitive(PType::I32, Nullability::Nullable);
211 assert_eq!(
212 fill_null(root(), lit(0i32)).return_dtype(&dtype).unwrap(),
213 DType::Primitive(PType::I32, Nullability::NonNullable)
214 );
215 }
216
217 #[test]
218 fn replace_children() {
219 let expr = fill_null(root(), lit(0i32));
220 expr.with_children(vec![root(), lit(0i32)])
221 .vortex_expect("operation should succeed in test");
222 }
223
224 #[test]
225 fn evaluate() {
226 let mut ctx = array_session().create_execution_ctx();
227 let test_array =
228 PrimitiveArray::from_option_iter([Some(1i32), None, Some(3), None, Some(5)])
229 .into_array();
230
231 let expr = fill_null(root(), lit(42i32));
232 let result = test_array.apply(&expr).unwrap();
233
234 assert_eq!(
235 result.dtype(),
236 &DType::Primitive(PType::I32, Nullability::NonNullable)
237 );
238 assert_arrays_eq!(
239 result,
240 PrimitiveArray::from_iter([1i32, 42, 3, 42, 5]),
241 &mut ctx
242 );
243 }
244
245 #[test]
246 fn evaluate_struct_field() {
247 let mut ctx = array_session().create_execution_ctx();
248 let test_array = StructArray::from_fields(&[(
249 "a",
250 PrimitiveArray::from_option_iter([Some(1i32), None, Some(3)]).into_array(),
251 )])
252 .unwrap()
253 .into_array();
254
255 let expr = fill_null(get_item("a", root()), lit(0i32));
256 let result = test_array.apply(&expr).unwrap();
257
258 assert_eq!(
259 result.dtype(),
260 &DType::Primitive(PType::I32, Nullability::NonNullable)
261 );
262 assert_arrays_eq!(result, PrimitiveArray::from_iter([1i32, 0, 3]), &mut ctx);
263 }
264
265 #[test]
266 fn evaluate_non_nullable_input() {
267 let mut ctx = array_session().create_execution_ctx();
268 let test_array = buffer![1i32, 2, 3].into_array();
269 let expr = fill_null(root(), lit(0i32));
270 let result = test_array.apply(&expr).unwrap();
271 assert_arrays_eq!(result, PrimitiveArray::from_iter([1i32, 2, 3]), &mut ctx);
272 }
273
274 #[test]
275 fn test_display() {
276 let expr = fill_null(get_item("value", root()), lit(0i32));
277 assert_eq!(expr.to_string(), "vortex.fill_null($.value, 0i32)");
278 }
279}