[
https://issues.apache.org/jira/browse/FLINK-41037?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yaoxuan Wu updated FLINK-41037:
-------------------------------
Description:
{code:python}
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as D
t = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
t.create_temporary_view('l', t.from_elements([(None, 1.0)], D.ROW([D.FIELD('k',
D.STRING()), D.FIELD('x', D.DOUBLE())])))
t.create_temporary_view('r', t.from_elements([(None,)], D.ROW([D.FIELD('k',
D.STRING())])))
J = "SELECT g.k, g.s, r.k AS rk FROM (SELECT k, SUM(x) AS s FROM l GROUP BY k)
g FULL JOIN r ON g.k = r.k"
t.execute_sql(J).print()
# (NULL, 1.0, NULL), (NULL, NULL, NULL) correct: two distinct rows
t.execute_sql(f"SELECT DISTINCT * FROM ({J})").print()
# expected: (NULL, 1.0, NULL), (NULL, NULL, NULL)
# actual: (NULL, NULL, NULL)
{code}
The plan groups by {{k, rk}} and keeps {{s}} as {{auxGrouping}}, i.e. it
assumes {{k}} determines {{s}}. That holds for the GROUP BY output but not
after a FULL or RIGHT join, which pads the left side with NULLs: the real row
(NULL, 1.0) and the padded row (NULL, NULL) fall into one group. The assumption
comes from {{FlinkRelMdUniqueGroups.getUniqueGroups}}, whose fallback for a
Join ignores the join type. Streaming mode is correct.
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
was:{code:python} from pyflink.table import EnvironmentSettings,
TableEnvironment, DataTypes as D t =
TableEnvironment.create(EnvironmentSettings.in_batch_mode())
t.create_temporary_view('l', t.from_elements([(None, 1.0)], D.ROW([D.FIELD('k',
D.STRING()), D.FIELD('x', D.DOUBLE())]))) t.create_temporary_view('r',
t.from_elements([(None,)], D.ROW([D.FIELD('k', D.STRING())]))) J = "SELECT g.k,
g.s, r.k AS rk FROM (SELECT k, SUM(x) AS s FROM l GROUP BY k) g FULL JOIN r ON
g.k = r.k" t.execute_sql(J).print() # (NULL, 1.0, NULL), (NULL, NULL, NULL)
correct: two distinct rows t.execute_sql(f"SELECT DISTINCT * FROM
({J})").print() # expected: (NULL, 1.0, NULL), (NULL, NULL, NULL) # actual:
(NULL, NULL, NULL) {code} The plan groups by {{k, rk}} and keeps {{s}} as
{{auxGrouping}}, i.e. it assumes {{k}} determines {{s}}. That holds for the
GROUP BY output but not after a FULL or RIGHT join, which pads the left side
with NULLs: the real row (NULL, 1.0) and the padded row (NULL, NULL) fall into
one group. The assumption comes from
{{FlinkRelMdUniqueGroups.getUniqueGroups}}, whose fallback for a Join ignores
the join type. Streaming mode is correct. Reproduces on 2.0.1, 2.2.1, 2.3.0
and master.
> DISTINCT over a FULL OUTER JOIN drops a row when the grouping key is NULL
> (batch mode)
> --------------------------------------------------------------------------------------
>
> Key: FLINK-41037
> URL: https://issues.apache.org/jira/browse/FLINK-41037
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0
> Reporter: Yaoxuan Wu
> Priority: Major
>
> {code:python}
> from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as
> D
> t = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
> t.create_temporary_view('l', t.from_elements([(None, 1.0)],
> D.ROW([D.FIELD('k', D.STRING()), D.FIELD('x', D.DOUBLE())])))
> t.create_temporary_view('r', t.from_elements([(None,)], D.ROW([D.FIELD('k',
> D.STRING())])))
> J = "SELECT g.k, g.s, r.k AS rk FROM (SELECT k, SUM(x) AS s FROM l GROUP BY
> k) g FULL JOIN r ON g.k = r.k"
> t.execute_sql(J).print()
> # (NULL, 1.0, NULL), (NULL, NULL, NULL) correct: two distinct rows
> t.execute_sql(f"SELECT DISTINCT * FROM ({J})").print()
> # expected: (NULL, 1.0, NULL), (NULL, NULL, NULL)
> # actual: (NULL, NULL, NULL)
> {code}
> The plan groups by {{k, rk}} and keeps {{s}} as {{auxGrouping}}, i.e. it
> assumes {{k}} determines {{s}}. That holds for the GROUP BY output but not
> after a FULL or RIGHT join, which pads the left side with NULLs: the real row
> (NULL, 1.0) and the padded row (NULL, NULL) fall into one group. The
> assumption comes from {{FlinkRelMdUniqueGroups.getUniqueGroups}}, whose
> fallback for a Join ignores the join type. Streaming mode is correct.
> Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
