chore: use datafusion impl of bit_count function (#3616)

* impl map_from_entries

* Revert "impl map_from_entries"

This reverts commit 768b3e9.

* Use datafusion impl of bit_count

* fix fmt

* Resolve conflicts

---------

Co-authored-by: Kazantsev Maksim <mn.kazantsev@gmail.com>

Author: kazantsev-maksim

native/core/src/execution/jni_api.rs

@@ -40,6 +40,7 @@ use datafusion::{
     prelude::{SessionConfig, SessionContext},
 };
 use datafusion_comet_proto::spark_operator::Operator;
+use datafusion_spark::function::bitwise::bit_count::SparkBitCount;
 use datafusion_spark::function::bitwise::bit_get::SparkBitGet;
 use datafusion_spark::function::bitwise::bitwise_not::SparkBitwiseNot;
 use datafusion_spark::function::datetime::date_add::SparkDateAdd;
@@ -55,6 +56,7 @@ use datafusion_spark::function::math::hex::SparkHex;
 use datafusion_spark::function::math::width_bucket::SparkWidthBucket;
 use datafusion_spark::function::string::char::CharFunc;
 use datafusion_spark::function::string::concat::SparkConcat;
+use datafusion_spark::function::string::space::SparkSpace;
 use futures::poll;
 use futures::stream::StreamExt;
 use jni::objects::JByteBuffer;
@@ -401,6 +403,7 @@ fn register_datafusion_spark_function(session_ctx: &SessionContext) {
     session_ctx.register_udf(ScalarUDF::new_from_impl(MapFromEntries::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkCrc32::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkSpace::default()));
+    session_ctx.register_udf(ScalarUDF::new_from_impl(SparkBitCount::default()));
 }
 
 /// Prepares arrow arrays for output.
@@ -911,7 +914,6 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_logMemoryUsage(
 
 use crate::execution::columnar_to_row::ColumnarToRowContext;
 use arrow::ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema};
-use datafusion_spark::function::string::space::SparkSpace;
 
 /// Initialize a native columnar to row converter.
 ///

native/spark-expr/src/bitwise_funcs/bitwise_count.rs

@@ -22,8 +22,8 @@ use crate::math_funcs::modulo_expr::spark_modulo;
 use crate::{
     spark_ceil, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_isnan,
     spark_lpad, spark_make_decimal, spark_read_side_padding, spark_round, spark_rpad, spark_unhex,
-    spark_unscaled_value, EvalMode, SparkBitwiseCount, SparkContains, SparkDateDiff,
-    SparkDateTrunc, SparkMakeDate, SparkSizeFunc,
+    spark_unscaled_value, EvalMode, SparkContains, SparkDateDiff, SparkDateTrunc, SparkMakeDate,
+    SparkSizeFunc,
 };
 use arrow::datatypes::DataType;
 use datafusion::common::{DataFusionError, Result as DataFusionResult};
@@ -191,7 +191,6 @@ pub fn create_comet_physical_fun_with_eval_mode(
 
 fn all_scalar_functions() -> Vec<Arc<ScalarUDF>> {
     vec![
-        Arc::new(ScalarUDF::new_from_impl(SparkBitwiseCount::default())),
         Arc::new(ScalarUDF::new_from_impl(SparkContains::default())),
         Arc::new(ScalarUDF::new_from_impl(SparkDateDiff::default())),
         Arc::new(ScalarUDF::new_from_impl(SparkDateTrunc::default())),

native/spark-expr/src/bitwise_funcs/mod.rs

@@ -41,7 +41,6 @@ pub use predicate_funcs::{spark_isnan, RLike};
 
 mod agg_funcs;
 mod array_funcs;
-mod bitwise_funcs;
 mod comet_scalar_funcs;
 pub mod hash_funcs;
 
@@ -61,7 +60,6 @@ mod math_funcs;
 mod nondetermenistic_funcs;
 
 pub use array_funcs::*;
-pub use bitwise_funcs::*;
 pub use conditional_funcs::*;
 pub use conversion_funcs::*;
 pub use nondetermenistic_funcs::*;

native/spark-expr/src/comet_scalar_funcs.rs

@@ -20,7 +20,7 @@
 package org.apache.comet.serde
 
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.types.{ByteType, IntegerType, LongType}
+import org.apache.spark.sql.types.{ByteType, LongType}
 
 import org.apache.comet.serde.QueryPlanSerde._
 
@@ -140,14 +140,4 @@ object CometBitwiseGet extends CometExpressionSerde[BitwiseGet] {
   }
 }
 
-object CometBitwiseCount extends CometExpressionSerde[BitwiseCount] {
-  override def convert(
-      expr: BitwiseCount,
-      inputs: Seq[Attribute],
-      binding: Boolean): Option[ExprOuterClass.Expr] = {
-    val childProto = exprToProto(expr.child, inputs, binding)
-    val bitCountScalarExpr =
-      scalarFunctionExprToProtoWithReturnType("bit_count", IntegerType, false, childProto)
-    optExprWithInfo(bitCountScalarExpr, expr, expr.children: _*)
-  }
-}
+object CometBitwiseCount extends CometScalarFunction[BitwiseCount]("bit_count")