feat(velox): Support RESPECT NULLS for collect_list/collect_set

[yaooqinn]

What changes were proposed in this pull request?

Add ignoreNulls parameter to VeloxCollectList/VeloxCollectSet to support Spark's RESPECT NULLS syntax (SPARK-55256). When ignoreNulls=false, null elements are included in the collected array.

Changes:

• VeloxCollectSet/VeloxCollectList: Accept ignoreNulls parameter (default true for backward compatibility).
• CollectRewriteRule: Propagates ignoreNulls from Spark's CollectList/CollectSet using reflection, making it backward-compatible with Spark versions that don't have ignoreNulls.

Related:

• Velox PR: fix: RESPECT NULLS for Spark collect_list function facebookincubator/velox#16933 (collect_list RESPECT NULLS)
• Velox PR: feat(sparksql): Support RESPECT NULLS for Spark collect_set aggregate function facebookincubator/velox#16416 (collect_set RESPECT NULLS, merged)
• Spark JIRA: SPARK-55256

How was this patch tested?

CI validation. The default behavior (ignoreNulls=true) is unchanged, ensuring backward compatibility.

[zhouyuan]

@yaooqinn thanks, could you please also verify on a new veox branch with patches for collect_list/set - in the current branch the collect_set commit is actually reverted https://github.com/IBM/velox/commits/dft-2026_03_24/

[yaooqinn]

Attempted local verification with Velox patches (collect_set from upstream main + collect_list from PR #16933). The Velox native lib (libvelox.a) builds successfully with both patches. However, full Gluten build cannot complete locally due to pre-existing Scala version mismatch in cached Maven artifacts (unrelated to this PR).

The VeloxCollect changes are purely Scala/JVM-side (DeclarativeAggregate), so they're independent of the Velox native collect_set/collect_list functions. CI will provide the definitive verification.

@zhouyuan The native Velox patches for collect_set/collect_list add Aggregate::setConstantInputs() for the boolean ignoreNulls parameter, but since VeloxCollect uses Scala-based array operations, it doesn't need those native changes. The Gluten PR works with any Velox version.

[zhouyuan]

@yaooqinn not really
the patch (https://github.com/facebookincubator/velox/pull/16416/changes ) changed the signature of collect_set functions, which requires a modification on substrait layer

I just made a test with your patch + velox changes, the spark unit tests are failing

https://github.com/apache/gluten/actions/runs/23652189825/job/68903134168?pr=11712

E20260327 15:24:19.552174  1769 Exceptions.h:87] Line: /work/cpp/velox/substrait/SubstraitToVeloxPlan.cc:290, Function:toAggregationFunctionName, Expression: signatures.has_value() && signatures.value().size() > 0 Cannot find function signature for collect_set_merge_extract_array_row_VARCHAR_BIGINT_BIGINT_endrow in final aggregation step., Source: RUNTIME, ErrorCode: INVALID_STATE
15:24:19.555 WARN org.apache.spark.sql.execution.GlutenFallbackReporter: Validation failed for plan: SortAggregate[QueryId=405], due to: 
 - Native validation failed: 
   |- Validation failed due to exception caught at file:SubstraitToVeloxPlanValidator.cc line:1450 function:validate, thrown from file:SubstraitToVeloxPlan.cc line:290 function:toAggregationFunctionName, reason:Cannot find function signature for collect_set_merge_extract_array_row_VARCHAR_BIGINT_BIGINT_endrow in final aggregation step.

[yaooqinn]

Root Cause Analysis

Thanks @zhouyuan for catching this! I traced the issue:

What changed

Before PR #16416: collect_set had 1 signature → hasSameIntermediateTypesAcrossSignatures() returns false → companion function registered as collect_set_merge_extract (no suffix).

After PR #16416: collect_set has 2 signatures (1-arg + 2-arg with boolean), both with intermediateType("array(T)") → hasSameIntermediateTypesAcrossSignatures() returns true → companion function registered as collect_set_merge_extract_array_T (with suffix).

The mismatch

Gluten's SubstraitToVeloxPlan.cc::toAggregationFunctionName() (line 280-294):

1. First tries collect_set_merge_extract → not found (was registered with suffix now)
2. Falls through, constructs suffix from concrete result type: collect_set_merge_extract_array_row_VARCHAR_BIGINT_BIGINT_endrow
3. Looks up that name → not found (registered as generic collect_set_merge_extract_array_T)
4. Throws: Cannot find function signature

Fix options

1. In Velox: Ensure hasSameIntermediateTypesAcrossSignatures returns false by keeping companion functions registered without suffix. Could use ignoreDuplicates or separate registration for the 2-arg signature.
2. In Gluten Substrait C++: Update toAggregationFunctionName to try the generic (no-suffix) companion name via getAggregateFunctionSignatures(baseName + "_merge_extract") first with type resolution, similar to how Velox's own planner resolves generic companion functions.

I'll fix this in the Velox PR #16416 (or a follow-up) since the root cause is the signature change creating a suffix-based companion registration.

[rui-mo]

@yaooqinn I assume we should adapt to the signature change in Gluten (see toAggregationFunctionName which intends to resolve the name incompatibility caused by hasSameIntermediateTypesAcrossSignatures) rather than making additional changes in Velox.

[yaooqinn]

Pushed fix: Updated SubstraitToVeloxPlan.cc::toAggregationFunctionName() to handle generic-typed companion functions.

Change: After failing the concrete type suffix lookup (e.g., collect_set_merge_extract_array_row_VARCHAR_...), now falls back to getCompanionFunctionSignatures(baseName) API to discover the actual companion function names (e.g., collect_set_merge_extract_array_T).

The file compiles cleanly against the existing Velox EP. @zhouyuan could you verify with your test setup?