feat(spark): implement Spark bitmap function bitmap_count

[SparkApplicationMaster]

Which issue does this PR close?

part of #15914

Rationale for this change

Migrate spark functions from https://github.com/lakehq/sail/ to datafusion engine to unify codebase

What changes are included in this PR?

implement spark udf bitmap_count
https://spark.apache.org/docs/latest/api/sql/index.html#bitmap_count

Are these changes tested?

unit-tests and sqllogictests added

Are there any user-facing changes?

bitmap_count(binary_col) now can be called in queries

[Jefffrey]

Would this function belong under bitwise module perhaps, instead of misc?

https://github.com/apache/datafusion/tree/main/datafusion/spark/src/function/bitwise

[Jefffrey]

I think this needs to be specified in the signature?

e.g.

datafusion/datafusion/spark/src/function/bitwise/bit_count.rs

Lines 44 to 62 in f87763d

	impl SparkBitCount {
	pub fn new() -> Self {
	Self {
	signature: Signature::one_of(
	vec![
	TypeSignature::Exact(vec![DataType::Int8]),
	TypeSignature::Exact(vec![DataType::Int16]),
	TypeSignature::Exact(vec![DataType::Int32]),
	TypeSignature::Exact(vec![DataType::Int64]),
	TypeSignature::Exact(vec![DataType::UInt8]),
	TypeSignature::Exact(vec![DataType::UInt16]),
	TypeSignature::Exact(vec![DataType::UInt32]),
	TypeSignature::Exact(vec![DataType::UInt64]),
	],
	Volatility::Immutable,
	),
	}
	}
	}

[SparkApplicationMaster]

I tried, but the problem is I didn't understand how to provide TypeSignature for FixedSizeBinary
It requires some integer value on creation
If it can be achieved somehow, this would be really great!

[Jefffrey]

Good point; did a bit of looking and stumbled upon this:

datafusion/datafusion/expr-common/src/signature.rs

Lines 42 to 45 in a0248a9

	/// Constant that is used as a placeholder for any valid fixed size list.
	/// This is used where a function can accept a fixed size list type with any
	/// valid length. It exists to avoid the need to enumerate all possible fixed size list lengths.
	pub const FIXED_SIZE_LIST_WILDCARD: i32 = i32::MIN;

So I think you can use like so:

    pub fn new() -> Self {
        Self {
            signature: Signature::one_of(
                vec![
                    TypeSignature::Exact(vec![Binary]),
                    TypeSignature::Exact(vec![BinaryView]),
                    TypeSignature::Exact(vec![LargeBinary]),
                    TypeSignature::Exact(vec![FixedSizeBinary(FIXED_SIZE_LIST_WILDCARD)]),
                ],
                Volatility::Immutable,
            ),
        }
    }

Do note you'll need to export FIXED_SIZE_LIST_WILDCARD here like is done for TIMEZONE_WILDCARD so it can be imported:

datafusion/datafusion/expr/src/lib.rs

Lines 85 to 88 in a0248a9

	pub use datafusion_expr_common::signature::{
	ArrayFunctionArgument, ArrayFunctionSignature, Coercion, Signature, TypeSignature,
	TypeSignatureClass, Volatility, TIMEZONE_WILDCARD,
	};

And would be good to have a few test cases of different sizes for FixedSizeBinary to ensure this works, as well as some negative cases to ensure providing a type like Utf8 would fail as expected

[SparkApplicationMaster]

this doesn't work unfortunately

1. query failed: DataFusion error: Error during planning: Failed to coerce arguments to satisfy a call to 'bitmap_count' function: coercion from [FixedSizeBinary(2)] to the signature OneOf([Exact([Binary]), Exact([BinaryView]), Exact([LargeBinary]), Exact([FixedSizeBinary(-2147483648)])]) failed No function matches the given name and argument types 'bitmap_count(FixedSizeBinary(2))'. You might need to add explicit type casts.
        Candidate functions:
        bitmap_count(Binary)
        bitmap_count(BinaryView)
        bitmap_count(LargeBinary)
        bitmap_count(FixedSizeBinary(-2147483648))
[SQL] SELECT bitmap_count(arrow_cast(a, 'FixedSizeBinary(2)')) FROM (VALUES (X'1010'), (X'0AB0'), (X'FFFF'), (NULL)) AS t(a);
at /home/x/datafusion/datafusion/sqllogictest/test_files/spark/bitmap/bitmap_count.slt:55

tried also 0 and other consts
it only works when the input type fixed size equals to this const

I suppose adding a designated const and a branch to implement the coersion is needed here:

datafusion/datafusion/expr/src/type_coercion/functions.rs

Lines 877 to 889 in a0248a9

	// should be able to coerce wildcard fixed size list to non wildcard fixed size list
	(
	FixedSizeList(f_into, FIXED_SIZE_LIST_WILDCARD),
	FixedSizeList(f_from, size_from),
	) => match coerced_from(f_into.data_type(), f_from.data_type()) {
	Some(data_type) if &data_type != f_into.data_type() => {
	let new_field =
	Arc::new(f_into.as_ref().clone().with_data_type(data_type));
	Some(FixedSizeList(new_field, *size_from))
	}
	Some(_) => Some(FixedSizeList(Arc::clone(f_into), *size_from)),
	_ => None,
	},

[Jefffrey]

Ah my mistake, I saw FIXED_SIZE_LIST_WILDCARD and assumed it would work for FixedSizeBinary too 🤦

[Jefffrey]

I've raised #17291 to hopefully resolve this; for now can leave a TODO comment with explanation so we can track it

[SparkApplicationMaster]

@Jefffrey thanks for review!

in spark, bitmap functions are isolated in their own bitmap file and function group
https://github.com/apache/spark/blob/8530444e25b83971da4314c608aa7d763adeceb3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala#L842
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/bitmapExpressions.scala#L250
so maybe it's better to create this bitmap module in spark crate?
I'll modify the PR to implement this, so next bitmap functions can go where they belong

[Jefffrey]

Looks like we have some failing tests btw @SparkApplicationMaster

[SparkApplicationMaster]

missed recent changes about UDF hashing and equality, I suppose now the build should pass

[Jefffrey]

LGTM

[Jefffrey]

Thanks @SparkApplicationMaster