Support creating arrays with non-nullable elements in make_array #11923

Activity

1. Kimahriman

   added

   enhancementNew feature or requestNew feature or request

   on Aug 9, 2024
2. jayzhan211 commented on Aug 10, 2024

   

   jayzhan211

   on Aug 10, 2024

   Contributor

   More actions

   What might be the downside if the array is always nullable? Loss the possibility for non-null optimization?
   If there is such a need for most of the functions, we might not need return_type anymore but return_type_from_exprs instead. That is a huge change.
3. Kimahriman commented on Aug 10, 2024

   

   Kimahriman

   on Aug 10, 2024

   ContributorAuthor

   More actions

   What might be the downside if the array is always nullable? Loss the possibility for non-null optimization

   That would be the main thing. In this case I came across this trying to integrate array support into datafusion-comet and it was inconsistent with Spark's handling of arrays. Maybe not the best reason to change, but it also seems weird to have nullability be tracked but not track it correctly.

   If there is such a need for most of the functions, we might not need return_type anymore but return_type_from_exprs instead. That is a huge change.

   The actual change per function isn't that much, working on what the make_array update would look like.

   👀React with 👀1Reacted by Jay Zhan
4. Kimahriman commented on Aug 10, 2024

   

   Kimahriman

   on Aug 10, 2024

   ContributorAuthor

   More actions

   One of the downstream affects would be ever so slightly increased encoding/decoding of Parquet files with arrays of non-nullable elements
5. jayzhan211 commented on Aug 12, 2024

   

   jayzhan211

   on Aug 12, 2024

   Last edited by jayzhan211

   Contributor

   More actions

   @alamb, I think we could have a breaking change on return_type with ReturnTypeArgs, and remove return_type_from_exprs.

   fn return_type(&self, _args: ReturnTypeArgs) -> Result<DataType> {
   
   struct ReturnTypeArgs<'a> {
       args: &'a [Expr],
       schema: &'a dyn ExprSchema,
       arg_types: &'a [DataType],
   }

   Alternatively, we switch all the function from return_type to return_type_from_exprs, since I think most of them would expect the same nullable from input.
6. Kimahriman commented on Aug 12, 2024

   

   Kimahriman

   on Aug 12, 2024

   ContributorAuthor

   More actions

   After digging a little more it seems like there's some other issues that might make this significantly more difficult:

   • When invoke is called, there's no way to know the logical nullability of the ColumnarValue's/ArrayRef's being passed in. The only thing available is Array::is_nullable, but it looks like all that does it check if the current array has any nulls or not, so that could vary from batch to batch. So inside invoke there's no way to know if field inside

       Ok(Arc::new(GenericListArray::<O>::try_new(
           Arc::new(Field::new("item", data_type, true)),
           OffsetBuffer::new(offsets.into()),
           arrow_array::make_array(data),
           None,
       )?))

   should be logically nullable or not. And arrow will verify this when creating a record batch, which is the error I ran into initially with the nullability not matching: https://github.com/apache/arrow-rs/blob/5c5a94a11f01a286dd03b18af0f11c327a9accc6/arrow-array/src/record_batch.rs#L205

   • ScalarUDF's in general don't have a concept of nullability, they are always considered nullable. Only tangentially related but just a larger issue of nullability and ScalarUDFs: https://github.com/apache/datafusion/blob/main/datafusion/physical-expr/src/scalar_function.rs#L120
7. jayzhan211 commented on Aug 12, 2024

   

   jayzhan211

   on Aug 12, 2024

   Last edited by jayzhan211

   Contributor

   More actions

   We could have another invoke_with_schema function to get the nullable from schema. Of course, another incompatible change 😢

   We have schema here
   

   datafusion/datafusion/physical-expr/src/scalar_function.rs

   Lines 214 to 239 in 18193e6

   	pub fn create_physical_expr(
   	fun: &ScalarUDF,
   	input_phy_exprs: &[Arc<dyn PhysicalExpr>],
   	input_schema: &Schema,
   	args: &[Expr],
   	input_dfschema: &DFSchema,
   	) -> Result<Arc<dyn PhysicalExpr>> {
   	let input_expr_types = input_phy_exprs
   	.iter()
   	.map(|e| e.data_type(input_schema))
   	.collect::<Result<Vec<_>>>()?;
   	// verify that input data types is consistent with function's `TypeSignature`
   	data_types_with_scalar_udf(&input_expr_types, fun)?;
   	// Since we have arg_types, we dont need args and schema.
   	let return_type =
   	fun.return_type_from_exprs(args, input_dfschema, &input_expr_types)?;
   	Ok(Arc::new(ScalarFunctionExpr::new(
   	fun.name(),
   	Arc::new(fun.clone()),
   	input_phy_exprs.to_vec(),
   	return_type,
   	)))
   	}
8. jayzhan211 commented on Aug 12, 2024

   

   jayzhan211

   on Aug 12, 2024

   Last edited by jayzhan211

   Contributor

   More actions

   I remember there was a discussion about replacing ColumnarValue::Array(ArrayRef) with Recordbatch, which contains Schema

   #[derive(Clone, Debug, PartialEq)]
   pub struct RecordBatch {
       schema: SchemaRef,
       columns: Vec<Arc<dyn Array>>,
   
       /// The number of rows in this RecordBatch
       ///
       /// This is stored separately from the columns to handle the case of no columns
       row_count: usize,
   }

   Maybe Recordbatch is what we need for invoke 🤔