[ https://issues.apache.org/jira/browse/FLINK-20128?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17231253#comment-17231253 ]
Dian Fu commented on FLINK-20128: --------------------------------- [~fsk119] I'm still not digging into this issue. However, it seems that this job doesn't use PyFlink specific functionalities such as Python UDF/UDTF/UDAF, etc and PyFlink is only a language layer here which just translates the Python API calls to the Java ones and so I don't think this is a PyFlink's bug. > Data loss for over windows with rows unbounded preceding > -------------------------------------------------------- > > Key: FLINK-20128 > URL: https://issues.apache.org/jira/browse/FLINK-20128 > Project: Flink > Issue Type: Bug > Components: API / Python, Table SQL / Planner > Affects Versions: 1.12.0, 1.11.2 > Reporter: Thilo Schneider > Priority: Major > > When using partitioned, unbounded over windows, all but one partitions are > dropped in the output dataset: > {code:python} > # Setup > from pyflink.table import EnvironmentSettings, StreamTableEnvironment > from biafflink import debug_print_table > env_settings = > EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build() > t_env = StreamTableEnvironment.create( environment_settings=env_settings) > t_env.get_config().get_configuration().set_integer("table.exec.resource.default-parallelism", > 1) > t_env.execute_sql(""" > CREATE TABLE datagen ( > foo INT, > id AS mod(foo, 2), > message_time AS to_timestamp(from_unixtime(FLOOR(foo/2))), > WATERMARK FOR message_time AS message_time > ) WITH ( > 'connector' = 'datagen', > 'rows-per-second'='2', > 'fields.foo.kind'='sequence', > 'fields.foo.start'='0', > 'fields.foo.end'='19' > )""") > t_env.execute_sql("CREATE TABLE output (foo INT, id INT, lagfoo INT) WITH > ('connector' = 'print')") > {code} > Using bounded over windows, everything works as expected: > {code:python} > t = t_env.sql_query(""" > SELECT foo, id, avg(foo) OVER w AS lagfoo > FROM datagen > WINDOW w AS (PARTITION BY id ORDER BY message_time ROWS BETWEEN 1 > PRECEDING AND CURRENT ROW)""") > t.execute_insert("output") > {code} > yields > {code:python} > +I(0,0,0) > +I(1,1,1) > +I(2,0,1) > +I(3,1,2) > +I(4,0,3) > ... > {code} > If we change the window to unbounded preceding: > {code:python} > t = t_env.sql_query(""" > SELECT foo, id, avg(foo) OVER w AS lagfoo > FROM datagen > WINDOW w AS (PARTITION BY id ORDER BY message_time ROWS BETWEEN UNBOUNDED > PRECEDING AND CURRENT ROW)""") > t.execute_insert("output") > {code} > we loose all of id == 1: > {code:python} > +I(0,0,0) > +I(2,0,1) > +I(4,0,2) > +I(6,0,3) > +I(8,0,4) > ... > {code} > I observed this problem with various aggregate functions and both under > 1.11.2 and 1.12rc1. -- This message was sent by Atlassian Jira (v8.3.4#803005)