sepuri sai krishna created FLINK-41411:
------------------------------------------
Summary: GROUP BY, DISTINCT and COUNT(DISTINCT) on a BITMAP column
are rejected in batch but work in streaming
Key: FLINK-41411
URL: https://issues.apache.org/jira/browse/FLINK-41411
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner, Table SQL / Runtime
Affects Versions: 2.3.0
Reporter: sepuri sai krishna
A {{BITMAP}} grouping key is rejected in batch mode and accepted in streaming
mode.
The two binary literals below are two valid serializations of the same bitmap,
the set
{{\{1,2,3,4,5\}}}: 26 bytes as an array container and 15 bytes after
{{runOptimize()}}.
{code:sql}
CREATE TABLE src (k INT) WITH (
'connector' = 'datagen', 'number-of-rows' = '2',
'fields.k.kind' = 'sequence', 'fields.k.start' = '1', 'fields.k.end' = '2');
CREATE VIEW two AS SELECT BITMAP_FROM_BYTES(CASE WHEN k = 1
THEN X'3a30000001000000000004001000000001000200030004000500'
ELSE X'3b3000000100000400010001000400' END) AS bm FROM src;
SELECT BITMAP_CARDINALITY(bm), BITMAP_TO_STRING(bm) FROM two;
SELECT COUNT(*) FROM two GROUP BY bm;
SELECT COUNT(DISTINCT bm) FROM two;
SELECT l.bm = r.bm FROM two l, two r;
SELECT COUNT(*) FROM two GROUP BY BITMAP_TO_BYTES(bm);
{code}
||query||batch||streaming||
|{{BITMAP_CARDINALITY}}, {{BITMAP_TO_STRING}}|{{5, \{1,2,3,4,5\}}} twice|same|
|{{GROUP BY bm}}|rejected|one group of 2|
|{{COUNT(DISTINCT bm)}}|rejected|{{1}}|
|{{SELECT DISTINCT bm}}|rejected|one row|
|{{l.bm = r.bm}}|{{TRUE}}|{{TRUE}}|
|{{GROUP BY BITMAP_TO_BYTES(bm)}}|one group of 2|one group of 2|
In batch the three rejected statements fail with:
{code}
org.apache.flink.table.api.ValidationException: Type 'BITMAP NOT NULL' cannot
be ordered,
so it cannot be used as a key for sorting, grouping, or joining (for example in
ORDER BY,
GROUP BY, DISTINCT, or a join condition). Remove it from the key, or replace it
with a
value that can be ordered.
{code}
FLIP-556 gives {{BITMAP}} equality and hashing but no total order: "Equality
and hashing:
BITMAP supports equals() and hashCode(), enabling its use in GROUP BY and
DISTINCT" and
"Non-comparable: BITMAP does not define a total order."
Streaming matches that, and it is also right about equality: it puts the two
encodings in one
group and reports one distinct value.
Hashing is canonical by design. {{RoaringBitmapData#hashCode()}} and
{{#toBytes()}} call
{{runOptimize()}} first, and {{RoaringBitmapDataTest#testHashCode}} asserts
that two bitmaps
holding the same set hash equal through Flink's wrapper while the underlying
library's own
{{hashCode()}} differs for them. Grouping needs that hash, not an order.
Raising {{table.exec.resource.hash-agg.memory}} to 512 mb and setting
{{table.optimizer.agg-phase-strategy}} to {{ONE_PHASE}} does not change the
batch outcome, and
{{EXPLAIN}} of the same statement fails with the same error, so it is not a
cost-based choice
between aggregate strategies.
Grouping on the canonical bytes works in batch, which is the workaround:
{{SELECT COUNT(*) FROM two GROUP BY BITMAP_TO_BYTES(bm)}}.
A {{MAP}} grouping key is rejected in batch with the same message, see
FLINK-41328.
Reproduced on master, built from source, JDK 17. {{BITMAP}} was added in 2.3.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
