datafusion-contrib / datafusion-contrib/StreamFusion

Native window aggregation: support Flink split-distinct physical plans

Open
#166 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Java
Stars
120
Forks
8
Avg merge
7h 29m
Merged PRs (30d)
76

Description

Follow-up to the completed base window COUNT(DISTINCT) support in #97. On main aea6056352c402e0c7008f6f223c7574324d796a, enable table.optimizer.distinct-agg.split.enabled=true and execute this shape over a runtime source with event-time watermarks:

SELECT k, window_start, window_end, COUNT(DISTINCT s), SUM(v), COUNT(*)
FROM TABLE(TUMBLE(TABLE src, DESCRIPTOR(rt), INTERVAL '10' SECOND))
GROUP BY k, window_start, window_end;

TUMBLE, HOP and CUMULATE comparisons all match released Flink 2.2.1 through full fallback. Even with TWO_PHASE requested, every generated plan records:

Calc: unsupported function/operator: HASH_CODE
window aggregate: attached-window aggregation requires two-phase execution
global window aggregate: requires SUM/MIN/MAX/COUNT or paired AVG partials

The corresponding splitDistinct=false paths run native. The unchanged upstream WindowDistinctAggregateITCase contracts currently require fallback for all three splitDistinct=true methods.

Acceptance:

  • Support the actual optimized split-distinct stages and their attached/global accumulator layouts, in addition to the separately tracked HASH_CODE scalar.
  • Preserve NULLs, duplicates across buckets/partials, mixed ordinary aggregates, temporal bounds, watermarks and late rows.
  • Verify state cleanup and memory/RocksDB recovery for each admitted layout.
  • Require native work in every necessary stage of the three unchanged upstream methods and their backend variants.
  • Retain precise fallback for any unverified form and update the window coverage page.

Retracting DISTINCT remains tracked by #99; this issue does not duplicate that changelog extension.

HASH_CODE scalar tracking: #165.

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start from the unchanged upstream WindowDistinctAggregateITCase methods and the splitDistinct=true plans described in the issue, using commit aea6056352c402e0c7008f6f223c7574324d796a. Trace the attached-window and global aggregate layouts, then verify native execution across the listed backends and cases, with precise fallback for unverified forms and the window coverage page updated.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, sql
Domain
stream-processing
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.