Yaoxuan Wu created FLINK-41250:
----------------------------------
Summary: Wrong result with auto-parallelism: a KEEP_INPUT_AS_IS
exchange is replaced by a hash partitioner on duplicated keys [a, a]
Key: FLINK-41250
URL: https://issues.apache.org/jira/browse/FLINK-41250
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination, Table SQL / Planner
Affects Versions: 2.2.1, 2.3.0, 2.0.1
Reporter: Yaoxuan Wu
{code:python}
from pyflink.common import Configuration
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as D
c = Configuration()
for k, v in [('parallelism.default', '-1'),
('execution.batch.adaptive.auto-parallelism.enabled', 'true'),
('execution.batch.adaptive.auto-parallelism.min-parallelism', '2'),
('execution.batch.adaptive.auto-parallelism.max-parallelism',
'4')]:
c.set_string(k, v)
t =
TableEnvironment.create(EnvironmentSettings.new_instance().in_batch_mode().with_configuration(c).build())
t.create_temporary_view('t', t.from_elements(
[(1, True, None), (2, False, True), (3, False, True), (4, True, None), (5,
True, False),
(6, True, True), (7, True, False), (8, True, True), (9, True, False), (10,
True, False)],
D.ROW([D.FIELD('id', D.INT()), D.FIELD('a', D.BOOLEAN()), D.FIELD('b',
D.BOOLEAN())])))
rows = t.execute_sql("SELECT id, a IN (SELECT i.a FROM t i WHERE i.b IS NOT
DISTINCT FROM o.a) AS m FROM t o").collect()
print(sorted(tuple(r) for r in rows))
# expected (and with any fixed parallelism): m = TRUE for 8 rows, FALSE for ids
2 and 3
# actual: m = FALSE for all 10 rows
{code}
Correct with a fixed parallelism (1, 2 or 4), with auto-parallelism and
max-parallelism 1, and with {{=}} instead of {{IS NOT DISTINCT FROM}}.
The decorrelated plan has an inner join whose right input is kept as is but
declared as hashed on a duplicated key:
{code}
HashJoin(joinType=[InnerJoin], where=[(IS NOT DISTINCT FROM(b, a0) AND (a0 =
a))], ...)
:- Exchange(distribution=[hash[a]])
+- Exchange(distribution=[keep_input_as_is[hash[a, a]]])
+- HashAggregate(isMerge=[true], groupBy=[a], select=[a])
+- Exchange(distribution=[hash[a]])
{code}
Cause: {{BatchExecExchange}} translates the non-strict {{KEEP_INPUT_AS_IS}}
into {{ForwardForConsecutiveHashPartitioner(createHashPartitioner(hash[a,
a]))}}. With the adaptive batch scheduler the stream graph is dynamic, and
{{StreamingJobGraphGenerator.tryConvertPartitionerForDynamicGraph}} replaces
this non-chainable edge by {{getHashPartitioner()}}, i.e. a
{{BinaryHashPartitioner}} over the keys {{[a, a]}}. {{HashCodeGenerator}}
hashes {{[a, a]}} as {{hash(a) * HASH_SALT(1) + hash(a)}}, not {{hash(a)}}, so
the right input is distributed differently from the left input ({{hash[a]}}),
matching rows never meet, and the join is empty. With a fixed parallelism the
edge stays a forward edge and the data keeps its {{hash[a]}} placement. The
planner should not record (or accept) {{hash[a, a]}} for an input that is
hashed on {{[a]}}, e.g. by deduplicating the hash keys.
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
