[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17849209#comment-17849209
]
xuyang commented on FLINK-34380:
Hi, [~rovboyko] . +1 to fix the wrong order. I'll take a look for your pr later.
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17849151#comment-17849151
]
Roman Boyko commented on FLINK-34380:
-
Hi [~xu_shuai_] , [~xuyangzhong] !
I still think that messages order is the main problem which should be addressed
in this issue.
Could you please review my comment above?
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17846895#comment-17846895
]
Roman Boyko commented on FLINK-34380:
-
Hi [~xu_shuai_] !
I can't agree with you because the order of messages for one key can't be
violated even for distributed system. Otherwise it will lead to inconsistent
result. You can check it by running the XuYang's example without last -D record
for both join sides:
{code:java}
val rows = Seq(
changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
// changelogRow("-D", java.lang.Integer.valueOf(1), "99")
) {code}
And the result will be:
{code:java}
++-++-++
| op | a | b | a0 |
b0 |
++-++-++
| +U | 1 | 1 | 1 |
99 |
| +U | 1 | 99 | 1 |
99 |
| -U | 1 | 1 | 1 |
99 |
++-++-++
{code}
This means that sink or downstream operator will receive the -U (retract
record) as last record. And I guess no downstream or sink operator can
correctly process elements in such order.
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17846626#comment-17846626
]
Shuai Xu commented on FLINK-34380:
--
Hi [~rovboyko] , sorry for late reply.
For the incorrect order of output records, the minibatch optimization is
designed to guanrantee final consistency. And the fix you mentioned has been
considered when the pr was reviewed. Flink is a distributed realtime processing
system. The order of output could be guanranteed on a node by using
LinkedHashMap, however, it could not be guranteed when join operator runs on
multiple nodes. So I think it makes little sense to keep the order here.
For the Rowkind, it was also reviewed. As you mentioned, it is a common problem
of MiniBatch functionality. It does not influence final result. From the
benefit perspective, this problem could be tolerable.
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17846520#comment-17846520
]
Roman Boyko commented on FLINK-34380:
-
Hi [~xuyangzhong] ! Thank you for your reply.
Yes, you're right - the RowKind still not fixed in this example. But I think we
should consider to fix the RowKind in separate issue because:
1) Incorrect RowKind in your example is the common problem of MiniBatch
functionality. It happens every time when +I and -U records are assigned to
first batch and then +U record is assigned to second batch. And it can't be
fixed easily and only for Join operator - we should try to reproduce the same
for Aggregate operator as well
2) While incorrect RowKind is not so serious problem, the incorrect order of
output records might be really critical because it leads to incorrect result
So I sugest to fix only incorrect order in this issue and create the separate
one for incorrect RowKind.
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17846516#comment-17846516
]
Roman Boyko commented on FLINK-34380:
-
Hi [~xuyangzhong] ! Thank you for your reply.
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17846485#comment-17846485
]
xuyang commented on FLINK-34380:
Sorry for this late reply. This commit for fix seems great. [~xu_shuai_] Can
you take a look to verify it again？
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17846141#comment-17846141
]
Roman Boyko commented on FLINK-34380:
-
[~xu_shuai_] , [~xuyangzhong] , could you please take a look?
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17842243#comment-17842243
]
Roman Boyko commented on FLINK-34380:
-
[~xuyangzhong] , [~xu_shuai_] , what do you think? May I create a PR for fixing
it?
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17839474#comment-17839474
]
Roman Boyko commented on FLINK-34380:
-
Hi [~xuyangzhong] , [~xu_shuai_] !
1) The RowKind can't be fixed in current architecture, because +I and +U are
separated in different batches in this example. And this would be are bit
tricky to fix it.
2) But the records order is really incorrect in this example and it can be
easily fixed - https://github.com/rovboyko/flink/tree/fix/FLINK-34380
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.20.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17822433#comment-17822433
]
Shuai Xu commented on FLINK-34380:
--
Let me take a look.
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.19.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-34380) Strange RowKind and records about intermediate output when using minibatch join
[
https://issues.apache.org/jira/browse/FLINK-34380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17822401#comment-17822401
]
xuyang commented on FLINK-34380:
Hi, [~xu_shuai_] . Can you help check it again?
> Strange RowKind and records about intermediate output when using minibatch
> join
> ---
>
> Key: FLINK-34380
> URL: https://issues.apache.org/jira/browse/FLINK-34380
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.19.0
>Reporter: xuyang
>Priority: Major
> Fix For: 1.19.0
>
>
> {code:java}
> // Add it in CalcItCase
> @Test
> def test(): Unit = {
> env.setParallelism(1)
> val rows = Seq(
> changelogRow("+I", java.lang.Integer.valueOf(1), "1"),
> changelogRow("-U", java.lang.Integer.valueOf(1), "1"),
> changelogRow("+U", java.lang.Integer.valueOf(1), "99"),
> changelogRow("-D", java.lang.Integer.valueOf(1), "99")
> )
> val dataId = TestValuesTableFactory.registerData(rows)
> val ddl =
> s"""
> |CREATE TABLE t1 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl)
> val ddl2 =
> s"""
> |CREATE TABLE t2 (
> | a int,
> | b string
> |) WITH (
> | 'connector' = 'values',
> | 'data-id' = '$dataId',
> | 'bounded' = 'false'
> |)
>""".stripMargin
> tEnv.executeSql(ddl2)
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED,
> Boolean.box(true))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
> Duration.ofSeconds(5))
> tEnv.getConfig.getConfiguration
> .set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, Long.box(3L))
> println(tEnv.sqlQuery("SELECT * from t1 join t2 on t1.a =
> t2.a").explain())
> tEnv.executeSql("SELECT * from t1 join t2 on t1.a = t2.a").print()
> } {code}
> Output:
> {code:java}
> ++-+-+-+-+
> | op | a | b | a0 | b0 |
> ++-+-+-+-+
> | +U | 1 | 1 | 1 | 99 |
> | +U | 1 | 99 | 1 | 99 |
> | -U | 1 | 1 | 1 | 99 |
> | -D | 1 | 99 | 1 | 99 |
> ++-+-+-+-+{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
