Verified on PR #220 after rebasing onto merged #222: commit 16ebb1e5a45cc5420abf90473b785d6c0b908a82, released Apache Flink 2.2.1, JDK 17, default compatibility settings. Each probe uses independent stock-Flink and native-enabled environments with bounded runtime DataStream input. Successful comparisons check output values and resolved result types before classifying execution.
Suggested priority: P2. This is a verified native support gap; the successful probes return equal results through explicit Flink fallback.
AVG(DISTINCT) remains unsupported for exact numeric inputs. Other distinct aggregates admit a single-phase plan but reject the local/global plan. These are separate from the window split-distinct operator in #166.
Runtime probe
Input type:
ROW<`f0` INT, `f1` TINYINT, `f2` SMALLINT, `f3` BIGINT, `f4` DECIMAL(20, 2), `f5` STRING, `f6` DATE, `f7` TIME(3), `f8` TIMESTAMP_LTZ(3), `f9` BOOLEAN> NOT NULL
Input: [+I[1, 3, 4, 5, 6.25, text, 2024-02-29, 12:34, 2024-02-29T12:34:56Z, true]]
SELECT f0, AVG(DISTINCT f4) FROM src GROUP BY f0;
Flink result: [[1, 6.250000]]; resolved type: [INT, DECIMAL(38, 6)]. Native substitutions: 0.
GROUP BY: DISTINCT is native for COUNT/SUM/MIN/MAX only
Runtime probe
Input type:
ROW<`f0` INT, `f1` TINYINT, `f2` SMALLINT, `f3` BIGINT, `f4` DECIMAL(20, 2), `f5` STRING, `f6` DATE, `f7` TIME(3), `f8` TIMESTAMP_LTZ(3), `f9` BOOLEAN> NOT NULL
Input: [+I[1, 3, 4, 5, 6.25, text, 2024-02-29, 12:34, 2024-02-29T12:34:56Z, true]]
SELECT f0, COUNT(DISTINCT f8) FROM src GROUP BY f0;
Flink result: [[1, 1]]; resolved type: [INT, BIGINT NOT NULL]. Native substitutions: 0.
local group aggregate: needs SUM/MIN/MAX/COUNT over bigint/int/double values with no widening of the partial, or AVG over any AvgAggFunction numeric, and bigint/int/string/boolean/date/timestamp/decimal grouping keys
global group aggregate: distinct merges are COUNT (over set-carriable value types) and SUM (over bigint/int)
Verified checklist:
- AVG(DISTINCT TINYINT/BIGINT/DECIMAL) falls back in one-phase and two-phase configurations.
- FIRST_VALUE(DISTINCT STRING) and LAST_VALUE(DISTINCT STRING) fall back.
- SUM(DISTINCT DECIMAL), COUNT(DISTINCT TIMESTAMP_LTZ) and COUNT(DISTINCT BOOLEAN) pass natively in one phase but fall back with TWO_PHASE plus mini-batching.
- SINGLE_VALUE(STRING) falls back in the selected local/global plan.
Two-phase settings: table.optimizer.agg-phase-strategy=TWO_PHASE, table.exec.mini-batch.enabled=true, table.exec.mini-batch.size=5, table.exec.mini-batch.allow-latency=100 ms. Preserve multiplicities, duplicate removal/retractions, NULLs, exact decimal scale and state restore. A configuration request alone is not proof of a local/global plan; assert the observed operators.
Acceptance
Preserve released Flink values, resolved types, NULL behavior and relevant error/short-circuit semantics. Add runtime-source SQL regressions that assert the specific native operator as well as result parity; do not use constant folding as proof of native support. Keep safe fallback for unverified forms and update the corresponding operator coverage documentation.
Verified on PR #220 after rebasing onto merged #222: commit
16ebb1e5a45cc5420abf90473b785d6c0b908a82, released Apache Flink 2.2.1, JDK 17, default compatibility settings. Each probe uses independent stock-Flink and native-enabled environments with bounded runtime DataStream input. Successful comparisons check output values and resolved result types before classifying execution.Suggested priority: P2. This is a verified native support gap; the successful probes return equal results through explicit Flink fallback.
AVG(DISTINCT) remains unsupported for exact numeric inputs. Other distinct aggregates admit a single-phase plan but reject the local/global plan. These are separate from the window split-distinct operator in #166.
Runtime probe
Input type:
Input:
[+I[1, 3, 4, 5, 6.25, text, 2024-02-29, 12:34, 2024-02-29T12:34:56Z, true]]Flink result:
[[1, 6.250000]]; resolved type:[INT, DECIMAL(38, 6)]. Native substitutions:0.Runtime probe
Input type:
Input:
[+I[1, 3, 4, 5, 6.25, text, 2024-02-29, 12:34, 2024-02-29T12:34:56Z, true]]Flink result:
[[1, 1]]; resolved type:[INT, BIGINT NOT NULL]. Native substitutions:0.Verified checklist:
Two-phase settings:
table.optimizer.agg-phase-strategy=TWO_PHASE,table.exec.mini-batch.enabled=true,table.exec.mini-batch.size=5,table.exec.mini-batch.allow-latency=100 ms. Preserve multiplicities, duplicate removal/retractions, NULLs, exact decimal scale and state restore. A configuration request alone is not proof of a local/global plan; assert the observed operators.Acceptance
Preserve released Flink values, resolved types, NULL behavior and relevant error/short-circuit semantics. Add runtime-source SQL regressions that assert the specific native operator as well as result parity; do not use constant folding as proof of native support. Keep safe fallback for unverified forms and update the corresponding operator coverage documentation.