[analytics-engine] Wire dc/earliest/latest/nth_value to DataFusion built-ins

[songkant-aws]

Description

Wires PPL dc/distinct_count/distinct_count_approx, earliest/latest, and nth_value to DataFusion built-ins on the analytics-engine route.

Backend rewrite via a new per-function WindowFunctionAdapter SPI (parallel to ScalarFunctionAdapter). DataFusion plugin registers three rewrites that BackendPlanAdapter applies before substrait emission:

• ARG_MIN(value, ts) → FIRST_VALUE(value) ORDER BY ts ASC (PPL earliest())
• ARG_MAX(value, ts) → LAST_VALUE(value) ORDER BY ts ASC (PPL latest())
• DISTINCT_COUNT_APPROX(x) → APPROX_COUNT_DISTINCT(x) (PPL dc() / distinct_count() / distinct_count_approx())

A small Rust wrapper UDAF aliases substrait's approx_count_distinct name to DataFusion's built-in approx_distinct so end-to-end execution works.

Unit-level coverage in WindowFunctionAdaptersTests pins the rewrite contracts. End-to-end IT in EventstatsCommandIT and StreamstatsCommandIT (8 dc methods, plus single-shard and multi-shard variants for dc / dc by) assert exact rows.

Why this PR depends on opensearch-project/sql#5466

PPLFuncImpTable.INSTANCE is a JVM-process-level singleton with a mutable aggExternalFunctionRegistry (ConcurrentHashMap). On current sql-plugin main, DISTINCT_COUNT_APPROX is registered into that map only as a side effect of constructing OpenSearchExecutionEngine — which is Guice-bound lazily and only ever instantiated when the V3 PPL path is exercised. Because the singleton registry is shared across paths once populated, the registration does become visible to the unified-query path eventually — but only if the running process has serviced at least one V3 request first.

Our IT environment only sends requests through /_analytics/ppl (RestUnifiedQueryAction), never through V3, so OpenSearchExecutionEngine is never constructed and DISTINCT_COUNT_APPROX is never registered. PPL parser then throws "Cannot resolve function: DISTINCT_COUNT_APPROX" before our backend rewrite gets a chance to run. opensearch-project/sql#5466 moves the registration into PPLFuncImpTable.populate() (class-init self-registration) using a logical marker UDAF, eliminating the V3-path precondition entirely.

Related Issues

• Depends on Register DISTINCT_COUNT_APPROX logical marker in PPLFuncImpTable sql#5466 for the dc-IT path (positive assertRowsEqual instead of negative assertErrorContains).

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 25c7497)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Fallback Logic Gap ArgFunctionAdapter.adapt() falls back to a passthrough rebuild when operands.size() != 2, but this silently produces incorrect output. If ARG_MIN/ARG_MAX arrive with 1 or 3+ operands due to a planner bug or future PPL syntax change, the rewritten RexOver will carry the wrong operator (ARG_MIN/ARG_MAX instead of FIRST_VALUE/LAST_VALUE) into substrait emission, likely causing a runtime error or wrong results. Consider logging a warning or throwing an exception instead of silent passthrough. if (operands.size() != 2) { // Unexpected shape — preserve original behavior as best as possible. return rebuild(original, original.getAggOperator(), operands, partitions, orderKeys, original.isDistinct(), cluster); } Filter Predicate Fragility The anonymous WindowFunctionConverter override filters out APPROX_COUNT_DISTINCT from the inherited sig list via a stream operation that reconstructs an ImmutableList. If getSigs() is called multiple times (e.g., during repeated conversions in a loop or retry scenario), the filter runs each time but the parent's sig list is already immutable, so the filter is safe. However, if a future refactor changes getSigs() to return a mutable list or caches the filtered result incorrectly, the filter could be bypassed. The current code is correct but the pattern is subtle. Consider caching the filtered list in a final field initialized in the constructor for clarity. @Override protected ImmutableList<FunctionMappings.Sig> getSigs() { return super.getSigs().stream() .filter(sig -> sig.operator != SqlStdOperatorTable.APPROX_COUNT_DISTINCT) .collect(ImmutableList.toImmutableList()); } };

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 25c7497

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Log warning for unexpected operand count When operands.size() != 2, the fallback path preserves the original operator but the operands list may have been adapted by scalar recursion. This creates an inconsistent state where adapted operands are paired with the original (potentially incompatible) operator. Consider logging a warning about the unexpected shape to aid debugging. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [66-88] private record ArgFunctionAdapter(SqlAggFunction target) implements WindowFunctionAdapter { + private static final Logger logger = LogManager.getLogger(ArgFunctionAdapter.class); + @Override public RexNode adapt( RexOver original, List<RexNode> operands, List<RexNode> partitions, List<RexFieldCollation> orderKeys, RelOptCluster cluster ) { if (operands.size() != 2) { - // Unexpected shape — preserve original behavior as best as possible. + logger.warn("Unexpected operand count {} for ARG_MIN/ARG_MAX, expected 2. Preserving original operator.", operands.size()); return rebuild(original, original.getAggOperator(), operands, partitions, orderKeys, original.isDistinct(), cluster); } return rebuild( original, target, List.of(operands.get(0)), partitions, appendAsc(orderKeys, operands.get(1)), original.isDistinct(), cluster ); } } Suggestion importance[1-10]: 6 __ Why: Adding logging for the unexpected operand count fallback path would aid debugging and make the fallback behavior more observable. The suggestion correctly identifies that the fallback preserves the original operator with adapted operands, which could be inconsistent. However, this is a defensive code path for malformed input, and the logging addition is a minor observability improvement rather than a correctness fix.	Low
	Return immutable list from appendAsc The appendAsc method creates a mutable ArrayList but returns it as List, allowing external modification. Since this list is used to construct immutable RexOver nodes via ImmutableList.copyOf(orderKeys) in the rebuild method, return an immutable list directly to prevent unintended mutations and clarify the contract. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [124-129] private static List<RexFieldCollation> appendAsc(List<RexFieldCollation> existing, RexNode key) { - List<RexFieldCollation> out = new ArrayList<>(existing.size() + 1); - out.addAll(existing); - out.add(new RexFieldCollation(key, Collections.emptySet())); - return out; + return ImmutableList.<RexFieldCollation>builderWithExpectedSize(existing.size() + 1) + .addAll(existing) + .add(new RexFieldCollation(key, Collections.emptySet())) + .build(); } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that appendAsc returns a mutable list that could be modified externally. However, the impact is limited since the list is immediately consumed by ImmutableList.copyOf() in rebuild(), preventing actual mutations from affecting the final RexOver. The improvement enhances code clarity and defensive programming but doesn't fix a critical bug.	Low

---

Previous suggestions

Suggestions up to commit 7b29502

Category	Suggestion	Impact
General	Log warning for unexpected operand count When ARG_MIN/ARG_MAX receives an unexpected operand count (not 2), the fallback silently preserves the original operator, which may cause downstream substrait emission to fail with cryptic errors. Log a warning with the actual operand count and operator name to aid debugging when this edge case occurs in production. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [75-78] if (operands.size() != 2) { // Unexpected shape — preserve original behavior as best as possible. + logger.warn("ARG_MIN/ARG_MAX adapter expected 2 operands but got {} for operator {}; falling back to passthrough", + operands.size(), original.getAggOperator().getName()); return rebuild(original, original.getAggOperator(), operands, partitions, orderKeys, original.isDistinct(), cluster); } Suggestion importance[1-10]: 6 __ Why: Adding a warning log for the unexpected operand count fallback case would help with debugging production issues. However, the suggestion assumes a logger exists but doesn't show its declaration, and logging alone doesn't change behavior or fix a bug.	Low
	Return immutable list from appendAsc The appendAsc method creates a mutable ArrayList but returns it as List, allowing external modification. Since this list is used to build immutable RexOver nodes via ImmutableList.copyOf(orderKeys) in the rebuild method, return an immutable list directly to prevent accidental mutation and clarify the contract. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [124-129] private static List<RexFieldCollation> appendAsc(List<RexFieldCollation> existing, RexNode key) { - List<RexFieldCollation> out = new ArrayList<>(existing.size() + 1); - out.addAll(existing); - out.add(new RexFieldCollation(key, Collections.emptySet())); - return out; + return ImmutableList.<RexFieldCollation>builder() + .addAll(existing) + .add(new RexFieldCollation(key, Collections.emptySet())) + .build(); } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that appendAsc returns a mutable list, but the impact is limited since rebuild already wraps it with ImmutableList.copyOf(orderKeys). The change improves code clarity and prevents potential misuse, but doesn't fix a bug.	Low
	Log when window operator is unrecognized If WindowFunction.resolveFunction returns null for an unrecognized window operator, the code silently skips adaptation. This may allow unsupported window functions to reach substrait emission and fail later. Consider logging a debug message when fn is null to trace which operators are not being adapted, aiding future debugging. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/BackendPlanAdapter.java [260-264] WindowFunction fn = WindowFunction.resolveFunction(over.getAggOperator()); +if (fn == null) { + LOGGER.debug("No WindowFunction mapping for operator {}, skipping adaptation", over.getAggOperator().getName()); +} WindowFunctionAdapter adapter = fn == null ? null : adapters.window().get(fn); if (adapter != null) { return adapter.adapt(over, adaptedOperands, adaptedPartitionKeys, adaptedOrderKeys, cluster); } Suggestion importance[1-10]: 5 __ Why: Adding a debug log when WindowFunction.resolveFunction returns null would aid debugging, but this is a low-impact observability improvement. The code already handles the null case correctly by skipping adaptation, and unrecognized operators would fail later with clear errors during substrait emission.	Low

Suggestions up to commit 2c731ef

Category	Suggestion	Impact
Possible issue	Validate adapter returns non-null RexNode The code retrieves a WindowFunctionAdapter from the map but does not validate that the adapter's returned RexNode is non-null. If a backend's adapter implementation returns null, the subsequent code will propagate this null reference, potentially causing a NullPointerException downstream. Add a null check after the adapter invocation to fail fast with a clear error message. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/BackendPlanAdapter.java [265-269] -private static RexNode adaptOver( - RexOver over, - Adapters adapters, - List<FieldStorageInfo> fieldStorage, - RelOptCluster cluster, - List<RexNode> adaptedOperands, - boolean operandsChanged -) { - ... - WindowFunction fn = WindowFunction.resolveFunction(over.getAggOperator()); - WindowFunctionAdapter adapter = fn == null ? null : adapters.window().get(fn); - if (adapter != null) { - return adapter.adapt(over, adaptedOperands, adaptedPartitionKeys, adaptedOrderKeys, cluster); +WindowFunction fn = WindowFunction.resolveFunction(over.getAggOperator()); +WindowFunctionAdapter adapter = fn == null ? null : adapters.window().get(fn); +if (adapter != null) { + RexNode adapted = adapter.adapt(over, adaptedOperands, adaptedPartitionKeys, adaptedOrderKeys, cluster); + if (adapted == null) { + throw new IllegalStateException("WindowFunctionAdapter for " + fn + " returned null"); } - ... + return adapted; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that if a WindowFunctionAdapter returns null, it could cause a NullPointerException downstream. Adding a null check with a clear error message would improve robustness and fail-fast behavior. However, the contract of WindowFunctionAdapter.adapt (line 52 in WindowFunctionAdapter.java) returns RexNode, and well-behaved implementations should not return null. The check is a defensive measure that improves error handling but addresses a scenario that should not occur under correct implementation.	Medium
General	Log warning on unexpected operand count When operands.size() != 2, the adapter falls back to rebuilding with the original operator, but this silently accepts malformed input without logging or signaling the unexpected condition. Consider logging a warning when the fallback path is taken so operators can detect and investigate queries that produce unexpected operand counts during development or debugging. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [92-95] private record ArgFunctionAdapter(SqlAggFunction target) implements WindowFunctionAdapter { @Override public RexNode adapt( RexOver original, List<RexNode> operands, List<RexNode> partitions, List<RexFieldCollation> orderKeys, RelOptCluster cluster ) { if (operands.size() != 2) { // Unexpected shape — preserve original behavior as best as possible. + LOGGER.warn("ARG_MIN/ARG_MAX adapter expected 2 operands but got {}, falling back to passthrough", operands.size()); return rebuild(original, original.getAggOperator(), operands, partitions, orderKeys, original.isDistinct(), cluster); } ... } } Suggestion importance[1-10]: 5 __ Why: Adding a warning log when the fallback path is taken would aid debugging and help detect malformed queries during development. However, the fallback is intentionally designed to preserve original behavior gracefully, and logging every unexpected operand count could add noise. The suggestion is valid but offers moderate value for maintainability and debugging.	Low
	Return immutable list from helper method The appendAsc method creates a mutable ArrayList and returns it as List, allowing external callers to modify the returned list. This could lead to unintended side effects if the caller mutates the list. Consider returning an immutable list using Collections.unmodifiableList() or List.copyOf() to prevent external modifications. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [141-146] private static List<RexFieldCollation> appendAsc(List<RexFieldCollation> existing, RexNode key) { List<RexFieldCollation> out = new ArrayList<>(existing.size() + 1); out.addAll(existing); out.add(new RexFieldCollation(key, Collections.emptySet())); - return out; + return Collections.unmodifiableList(out); } Suggestion importance[1-10]: 3 __ Why: The suggestion to return an immutable list is a defensive programming practice that prevents external modification. However, the appendAsc helper is private static and only called internally within ArgFunctionAdapter.adapt, where the returned list is immediately passed to rebuild and not stored or exposed. The risk of unintended mutation is minimal in this controlled context, so the improvement is marginal.	Low

Suggestions up to commit b851d9e

Category	Suggestion	Impact
General	Log warning for unexpected operand count The fallback path for unexpected operand counts silently preserves the original operator without logging or signaling the anomaly. Consider adding a warning log to alert operators when ARG_MIN/ARG_MAX receives an unexpected operand shape, as this may indicate a planner bug or unsupported query pattern that should be investigated. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [92-95] if (operands.size() != 2) { // Unexpected shape — preserve original behavior as best as possible. + // Log warning: ARG_MIN/ARG_MAX expects exactly 2 operands (value, ts) return rebuild(original, original.getAggOperator(), operands, partitions, orderKeys, original.isDistinct(), cluster); } Suggestion importance[1-10]: 5 __ Why: Adding a warning log when ARG_MIN/ARG_MAX receives unexpected operand counts would help diagnose planner bugs or unsupported query patterns. This is a reasonable observability improvement, though the suggestion only mentions adding a comment rather than actual logging code, which limits its immediate value.	Low
	Return immutable list from appendAsc The appendAsc method creates a mutable ArrayList and returns it, but callers may expect an immutable list since the method signature doesn't indicate mutability. Consider returning an immutable list using List.copyOf() or ImmutableList.copyOf() to prevent unintended modifications and align with the immutable pattern used elsewhere in the codebase (e.g., ImmutableList.copyOf(orderKeys) in rebuild). sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [141-146] private static List<RexFieldCollation> appendAsc(List<RexFieldCollation> existing, RexNode key) { List<RexFieldCollation> out = new ArrayList<>(existing.size() + 1); out.addAll(existing); out.add(new RexFieldCollation(key, Collections.emptySet())); - return out; + return List.copyOf(out); } Suggestion importance[1-10]: 4 __ Why: The suggestion to return an immutable list is valid for defensive programming, but the impact is low since appendAsc is a private helper method whose return value is immediately consumed by rebuild (which wraps it in ImmutableList.copyOf(orderKeys) anyway). The extra copy has minimal performance impact but adds a layer of safety.	Low
	Guard against null WindowFunction lookup When WindowFunction.resolveFunction returns null but an adapter exists in the map for a null key (theoretically possible if the map allows null keys), the code would skip adaptation. Add an explicit null-check guard or document that adapters.window() must never contain null keys to prevent silent failures in adapter dispatch. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/BackendPlanAdapter.java [265-269] WindowFunction fn = WindowFunction.resolveFunction(over.getAggOperator()); -WindowFunctionAdapter adapter = fn == null ? null : adapters.window().get(fn); +if (fn == null) { + // No recognized window function; skip adapter lookup + if (operandsChanged || windowChanged) { + return cluster.getRexBuilder().makeOver(...); + } + return over; +} +WindowFunctionAdapter adapter = adapters.window().get(fn); if (adapter != null) { return adapter.adapt(over, adaptedOperands, adaptedPartitionKeys, adaptedOrderKeys, cluster); } Suggestion importance[1-10]: 3 __ Why: The suggestion addresses a theoretical edge case where resolveFunction returns null. However, the existing code already handles this correctly by checking adapter != null before calling adapt. The suggested refactoring adds clarity but doesn't fix an actual bug, and the concern about null keys in the map is unlikely given the Map.of() construction pattern used throughout the codebase.	Low

Suggestions up to commit 31702ea

Category	Suggestion	Impact
General	Fail fast on unexpected operand count When ARG_MIN/ARG_MAX receive unexpected operand counts, the fallback silently preserves the original operator instead of failing fast. This masks configuration errors and may produce incorrect substrait. Consider logging a warning or throwing an exception to surface the unexpected shape. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [92-95] if (operands.size() != 2) { - // Unexpected shape — preserve original behavior as best as possible. - return rebuild(original, original.getAggOperator(), operands, partitions, orderKeys, original.isDistinct(), cluster); + throw new IllegalStateException( + "ARG_MIN/ARG_MAX expects exactly 2 operands (value, ts), got " + operands.size() + ); } Suggestion importance[1-10]: 7 __ Why: The current fallback silently preserves incorrect shapes, which could mask configuration errors. Failing fast with an exception would surface issues earlier, improving debuggability. However, the comment suggests the fallback is intentional for robustness.	Medium
	Fail fast on unsupported window function When WindowFunction.fromRexOver returns null (unsupported window function), the code silently falls through to the default rebuild path. This may allow unsupported functions to reach substrait emission. Consider throwing an exception when fn is null to fail fast on unsupported window functions. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/BackendPlanAdapter.java [265-269] WindowFunction fn = WindowFunction.fromRexOver(over); -WindowFunctionAdapter adapter = fn == null ? null : adapters.window().get(fn); +if (fn == null) { + throw new IllegalStateException( + "Unsupported window function: " + over.getAggOperator().getName() + ); +} +WindowFunctionAdapter adapter = adapters.window().get(fn); if (adapter != null) { return adapter.adapt(over, adaptedOperands, adaptedPartitionKeys, adaptedOrderKeys, cluster); } Suggestion importance[1-10]: 6 __ Why: When fromRexOver returns null, the code falls through to a default rebuild. Throwing an exception would catch unsupported functions earlier. However, the existing code may intentionally allow pass-through for functions without adapters, so this change could break valid use cases.	Low
	Return immutable list from helper The appendAsc method creates a mutable ArrayList but returns it as List, allowing external modification. Since this is a helper that builds order-key lists for window functions, return an immutable list to prevent accidental mutation after construction. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [141-146] private static List<RexFieldCollation> appendAsc(List<RexFieldCollation> existing, RexNode key) { List<RexFieldCollation> out = new ArrayList<>(existing.size() + 1); out.addAll(existing); out.add(new RexFieldCollation(key, Collections.emptySet())); - return out; + return Collections.unmodifiableList(out); } Suggestion importance[1-10]: 5 __ Why: Returning an immutable list prevents accidental mutation, which is a good defensive practice. However, the current code works correctly and the risk of external mutation is low in this context. The improvement is minor.	Low

Suggestions up to commit fdb5ba1

Category	Suggestion	Impact
General	Log warning on unexpected operand count When ARG_MIN/ARG_MAX receive unexpected operand counts, the fallback silently preserves the original operator instead of the target (FIRST_VALUE/LAST_VALUE). This may produce incorrect substrait emission. Consider logging a warning or throwing an exception to surface the unexpected shape. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [92-95] if (operands.size() != 2) { - // Unexpected shape — preserve original behavior as best as possible. + // Log unexpected shape before fallback + logger.warn("ARG_MIN/ARG_MAX adapter expected 2 operands, got {}. Falling back to original operator.", operands.size()); return rebuild(original, original.getAggOperator(), operands, partitions, orderKeys, original.isDistinct(), cluster); } Suggestion importance[1-10]: 6 __ Why: Adding logging for unexpected operand counts would help diagnose issues during development and debugging. However, the fallback behavior is already documented as "preserve original behavior as best as possible," and the suggestion introduces a dependency on a logger that isn't shown in the code.	Low
	Return immutable list from helper The appendAsc method creates a mutable ArrayList but returns it as List, allowing external modification. Consider returning an immutable list to prevent unintended mutations after the adapter completes. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/WindowFunctionAdapters.java [141-146] private static List<RexFieldCollation> appendAsc(List<RexFieldCollation> existing, RexNode key) { List<RexFieldCollation> out = new ArrayList<>(existing.size() + 1); out.addAll(existing); out.add(new RexFieldCollation(key, Collections.emptySet())); - return out; + return Collections.unmodifiableList(out); } Suggestion importance[1-10]: 4 __ Why: Returning an immutable list prevents external modification, which is a good defensive practice. However, the impact is low since the list is only used internally within the adapter and there's no evidence of mutation issues in the current codebase.	Low

[github-actions]

❌ Gradle check result for 5f49795: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit a14b109

[github-actions]

Persistent review updated to latest commit 691717e

[github-actions]

❌ Gradle check result for 691717e: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit fd2d105

[github-actions]

Persistent review updated to latest commit 3d43c88

[github-actions]

❌ Gradle check result for 3d43c88: null

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit fdb5ba1

[github-actions]

❌ Gradle check result for fdb5ba1: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[mch2]

can we test these with multi shard case? calcs/mapping.json - i have tried setting for clickbench tests here #21848 to add randomization 1-2 shard, we can do somethign similar here.

We'll likely have variance with dc() being approximate, but to ensure the queries succeed.

[songkant-aws]

Added multi-shard variants in 31702ea:

• testEventstatsDistinctCount_3shard and testEventstatsDistinctCountByCountry_3shard reuse the existing DATASET_MULTI (3 shards) — eventstats dc is broadcast pattern so the per-row expectations are identical to single-shard, and HLL is exact at calcs's 17-row scale.
• testStreamstatsDistinctCountByCountry_3shard collapses to per-partition finals via | stats max(dc_str3) by str0 | sort str0 since streamstats running values aren't deterministic when shard parallelism reorders input rows — same shape as testStreamstatsBy_3shard.

Didn't go with DatasetProvisioner.provision-level random shard count for this PR since #21848's randomization helper isn't merged yet; sticking with the existing per-test 3-shard pattern this file already uses.

[songkant-aws]

Sanity-checked datafusion 53.1's HLL with a standalone Rust harness running the real algorithm over 3 shards × 100 iterations — dc(str3) always returns 1 and dc(str0) always returns 3, with zero variance. The merge step is just register-wise max with a fixed seed, so multi-shard and single-shard paths produce bit-identical state, and for these two specific values the count output sits well clear of any rounding boundary.

[github-actions]

Persistent review updated to latest commit 31702ea

[github-actions]

✅ Gradle check result for 31702ea: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.46%. Comparing base (b186488) to head (7b29502).
⚠️ Report is 9 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21823      +/-   ##
============================================
- Coverage     73.50%   73.46%   -0.05%     
- Complexity    75546    75569      +23     
============================================
  Files          6034     6034              
  Lines        342661   342661              
  Branches      49294    49294              
============================================
- Hits         251879   251719     -160     
- Misses        70790    70958     +168     
+ Partials      19992    19984       -8     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[sandeshkr419]

I see you are doing 2 things:

Part 1: Wire dc/earliest/latest/nth_value: no comments for now, good to go!
Part 2: Unify reduce-stage subtree conversion: This part should simplify after my changes: #21775 (aiming to get this in today). I have introduced state-shipping which unifies all agg state as Binary IPC, so the schema never shifts between SINGLE and FINAL - this should simplify your code a lot.

[github-actions]

Persistent review updated to latest commit b851d9e

[github-actions]

✅ Gradle check result for b851d9e: SUCCESS

[songkant-aws]

@sandeshkr419 thanks for the review. Rebased on main — both #21775's reduce_eval Binary-IPC path and #21875's overrideExchangeType transformer are on this branch now. Those address PARTIAL/FINAL aggregate row-type drift at the rel layer for STATE_EXPANDING / opaque-state aggregates so the schema doesn't shift between SINGLE and FINAL.

Could you take another look?

[github-actions]

Persistent review updated to latest commit 2c731ef

[github-actions]

❌ Gradle check result for 2c731ef: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[sandeshkr419]

1 major comment on support of dc, rest looks fine.

While you are addressing this, can you also please concise up the comments in your changes to avoid overflowing intermittent design decisions.

[github-actions]

Persistent review updated to latest commit 7b29502

[github-actions]

✅ Gradle check result for 7b29502: SUCCESS

[github-actions]

Persistent review updated to latest commit 25c7497