How to implement scalar/aggregate functions with their own config/parameter?

[Yohahaha]

Background Info

When add first/last Spark aggregate function, @mbasmanova points out there are some inconsistent behaviors for scalar function and aggregate function to provide configureable implementation, see #4578 (comment)

Spark size function has spark.sql.legacy.sizeOfNull config to control its behavior, and Velox provide a QueryConfig for it https://facebookincubator.github.io/velox/configs.html#spark-legacy-size-of-null.

velox/velox/functions/sparksql/Size.cpp

Lines 30 to 35 in e1305d4

	template <typename TInput>
	FOLLY_ALWAYS_INLINE void initialize(
	const core::QueryConfig& config,
	const TInput* input /*input*/) {
	legacySizeOfNull_ = config.get<bool>("spark.legacy-size-of-null", true);
	}

Spark first aggregate function provide a ignoreNull parameter in SQL usage: select first(expr, ignoreNull) from tmp. However, Velox aggregate function does not have any config class to initialize itself compared with scalar function.

Solutions

For function parameters in SQL

Parameters will be parsed by parser as each function's own member, does not shared in session or query. These functions should provide different implementations to fulfill it, for example, select first(expr) as not_ignore_null_expr, first(expr, true) as ignore_null_expr from tmp, downstream engine which use Velox for accelerating should invoke first() and first_ignore_null() separately.

For function depends on config

Some functions depends on static/session config to exhibit different behavior, we should pass a config class in its initialize method. For example, Spark size function depends on spark.sql.legacy.sizeOfNull, Presto date_add function depends on session timezone.

However, we lack of the ability for pass session config to aggregate functions. Consider if someone wants add a new aggregate function depends on session timezone, how could we implement it?

CC @mbasmanova

[mbasmanova]

Parameters will be parsed by parser as each function's own member, does not shared in session or query.

@Yohahaha I didn't know Spark allows this. Curious if you could share some pointer to documentation and example queries. Assuming ignore-nulls is such parameter, I agree with you that Velox needs to provide 2 separate function implementations.

[Yohahaha]

For example, Gluten use substrait to communicate with Velox, when we parse last(x, ignoreNull) from substrait to Velox, we can choose last or last_ignore_null depends on ignoreNull value, but we have no way to set ignoreNull as Velox aggregate function's argument.

[mbasmanova]

This should be possible. When creating CallTypedExpr for the last(x, true), you can pass ConstantTypedExpr as the second parameter.

[Yohahaha]

I got your points, and find related Issue #3993 #4140, I will follow up with first/last aggregate function refactor, thanks!

[Yohahaha]

This plan suggests that 'last' function is called with 2 inputs: one is a column, another a constant. Is this the case?

Actually, last function does not accept 2 inputs, above plan is just a toString method's output for readability. last only accept 1 input child, and second argument in its constructor is not Expression, just a boolean variable. Spark aggregate function use updateExpressions to update group value like addRawInput/addSingleGroupRawInput, last aggregate functions's updateExpressions only compute child expression and use ignoreNull to decide if branch.
https://github.com/apache/spark/blob/87ccfc2402394d168a90c8df494482c7a1d6a552/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Last.scala#L90-L102

As for sort_array in #3993, its constructor has 2 arguments, base and ascendingOrder, are all Expression, so it could be called with 2 inputs.
https://github.com/apache/spark/blob/9e17731f46aa26348e69f42d1a96882186022b80/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala#L1032-L1047

So, I think for first/last like aggregate functions, Velox provides last(x) and last_ignore_null(x) are make sense.

[mbasmanova]

@Yohahaha Thank you for clarifying. Makes sense to me. Appreciate the detailed explanation. Thank you for digging deep.

[mbasmanova]

Some functions depends on static/session config to exhibit different behavior, we should pass a config class in its initialize method.

Makes sense to me. Do we have an example of an aggregate function that needs access to session config?

[Yohahaha]

I don't find such functions either, maybe I'm wrong.

[Yohahaha]

I see Presto has similar ignoreNull function, how do we handle these? https://prestodb.io/docs/current/functions/window.html#value-functions

[mbasmanova]

@Yohahaha This is part of WindowNode

class WindowNode : public PlanNode {
 public:
  ...

  struct Function {
    CallTypedExprPtr functionCall;
    Frame frame;
    bool ignoreNulls;

    folly::dynamic serialize() const;

    static Function deserialize(const folly::dynamic& obj);
  };

  /// @param windowColumnNames specifies the output column
  /// names for each window function column. So
  /// windowColumnNames.length() = windowFunctions.length().
  WindowNode(
      PlanNodeId id,
      std::vector<FieldAccessTypedExprPtr> partitionKeys,
      std::vector<FieldAccessTypedExprPtr> sortingKeys,
      std::vector<SortOrder> sortingOrders,
      std::vector<std::string> windowColumnNames,
      std::vector<Function> windowFunctions,
      PlanNodePtr source);

@aditi-pandit Aditi, would you confirm whether ignoreNulls flag is currently being respected or not?

[aditi-pandit]

@mbasmanova : ignoreNulls flag is not respected yet. This flag is only for value functions (first, last, nth, lead, lag) for windows. We have some PRs in flight for adding this behavior since we have got to lead/lag functions now.

[mbasmanova]

@aditi-pandit Aditi, thank you for clarifying. It would be nice to add this functionality.

CC: @zacw7 Zac, FYI, this a gap currently present in the implementation of window functions.