[jira] [Closed] (FLINK-30282) Logical type ROW lost inner field's nullability after convert to RelDataType
[
https://issues.apache.org/jira/browse/FLINK-30282?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan closed FLINK-30282.
-
Resolution: Not A Problem
This is a by-design behavior according to Calcite
https://issues.apache.org/jira/browse/CALCITE-2464
> Logical type ROW lost inner field's nullability after convert to RelDataType
>
>
> Key: FLINK-30282
> URL: https://issues.apache.org/jira/browse/FLINK-30282
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
>Affects Versions: 1.16.0, 1.16.1
>Reporter: Jane Chan
>Assignee: Jane Chan
>Priority: Major
> Labels: pull-request-available
>
> h3. Issue History
> This is not a new issue, FLINK-13604 has tracked it before, and FLINK-16344
> spared efforts to fix it (but did not tweak the ut case mentioned in
> FLINK-13604, i.e.
> SqlToOperationConverterTest#testCreateTableWithFullDataTypes). Nevertheless,
> the FunctionITCase added by FLINK-16344, which validates the fix, has been
> removed in FLINK-16377.
> h3. How to Reproduce
> c.c2 lost nullability
> {code:java}
> Flink SQL> create table dummy (a array not null, b array not null>, c row) with ('connector' = 'datagen');
> [INFO] Execute statement succeed.
> Flink SQL> desc dummy;
> +--++---+-++---+
> | name | type | null | key | extras | watermark |
> +--++---+-++---+
> | a | ARRAY | FALSE | | | |
> | b | ARRAY | TRUE | | | |
> | c | ROW<`c1` INT, `c2` DOUBLE> | TRUE | | | |
> +--++---+-++---+
> 3 rows in set
> {code}
> h3. Root Cause
> Two places are causing this problem in ExtendedSqlRowTypeNameSpec.
> 1. dt.deriveType should also pass dt's nullability as well. See
> [https://github.com/apache/flink/blob/fb27e6893506006b9a3b1ac3e9b878fb6cad061a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/ExtendedSqlRowTypeNameSpec.java#L159]
>
> 2. StructKind should be PEEK_FIELDS_NO_EXPAND instead of FULLY_QUALIFIED(see
> [https://github.com/apache/calcite/blob/main/core/src/main/java/org/apache/calcite/rel/type/StructKind.java]),
> so that FlinkTypeFactory#createTypeWithNullability will not fall back to
> super implement. See
> [https://github.com/apache/flink/blob/fb27e6893506006b9a3b1ac3e9b878fb6cad061a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala#L417]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-30282) Logical type ROW lost inner field's nullability after convert to RelDataType
[
https://issues.apache.org/jira/browse/FLINK-30282?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17730853#comment-17730853
]
Jane Chan commented on FLINK-30282:
---
I rechecked the discussion
[CALCITE-2464|https://issues.apache.org/jira/browse/CALCITE-2464] and assume
the current behavior is by design, and I'd like to close this issue.
> Logical type ROW lost inner field's nullability after convert to RelDataType
>
>
> Key: FLINK-30282
> URL: https://issues.apache.org/jira/browse/FLINK-30282
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
>Affects Versions: 1.16.0, 1.16.1
>Reporter: Jane Chan
>Assignee: Jane Chan
>Priority: Major
> Labels: pull-request-available
>
> h3. Issue History
> This is not a new issue, FLINK-13604 has tracked it before, and FLINK-16344
> spared efforts to fix it (but did not tweak the ut case mentioned in
> FLINK-13604, i.e.
> SqlToOperationConverterTest#testCreateTableWithFullDataTypes). Nevertheless,
> the FunctionITCase added by FLINK-16344, which validates the fix, has been
> removed in FLINK-16377.
> h3. How to Reproduce
> c.c2 lost nullability
> {code:java}
> Flink SQL> create table dummy (a array not null, b array not null>, c row) with ('connector' = 'datagen');
> [INFO] Execute statement succeed.
> Flink SQL> desc dummy;
> +--++---+-++---+
> | name | type | null | key | extras | watermark |
> +--++---+-++---+
> | a | ARRAY | FALSE | | | |
> | b | ARRAY | TRUE | | | |
> | c | ROW<`c1` INT, `c2` DOUBLE> | TRUE | | | |
> +--++---+-++---+
> 3 rows in set
> {code}
> h3. Root Cause
> Two places are causing this problem in ExtendedSqlRowTypeNameSpec.
> 1. dt.deriveType should also pass dt's nullability as well. See
> [https://github.com/apache/flink/blob/fb27e6893506006b9a3b1ac3e9b878fb6cad061a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/ExtendedSqlRowTypeNameSpec.java#L159]
>
> 2. StructKind should be PEEK_FIELDS_NO_EXPAND instead of FULLY_QUALIFIED(see
> [https://github.com/apache/calcite/blob/main/core/src/main/java/org/apache/calcite/rel/type/StructKind.java]),
> so that FlinkTypeFactory#createTypeWithNullability will not fall back to
> super implement. See
> [https://github.com/apache/flink/blob/fb27e6893506006b9a3b1ac3e9b878fb6cad061a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala#L417]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-31830) Coalesce on nested fields with different nullabilities will get wrong plan
[
https://issues.apache.org/jira/browse/FLINK-31830?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17730846#comment-17730846
]
Jane Chan edited comment on FLINK-31830 at 6/9/23 7:28 AM:
---
It takes some time to reason out the cause, and sorry for the late update.
h4. 1. Identify the problem
The issue has been reproduced using Flink release-1.14.6 (which depends on
calcite-1.26).
h4. 2. Why SQL gets a correct plan, while API doesn't
First, the resolved schema differs. You can verify this by the following code
snippet.
{code:scala}
@Test
def testSchema(): Unit = {
// create temporary table t1
val tableDescriptor = TableDescriptor.forConnector("datagen")
.schema(Schema.newBuilder
.column("id", DataTypes.INT.notNull)
.column("a", DataTypes.ROW(DataTypes.FIELD("np",
DataTypes.INT.notNull())).nullable)
.build)
.build
tableEnv.createTemporaryTable("t1", tableDescriptor)
// create temporary table t2
tableEnv.executeSql(
s"""
|create temporary table t2 (
| id int not null,
| a row
|) with (
| 'connector' = 'datagen'
|)
|""".stripMargin)
val catalogManager =
tableEnv.asInstanceOf[StreamTableEnvironmentImpl].getCatalogManager
val result1 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t1"))
val result2 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t2"))
println(result1.get().getResolvedSchema)
println(result2.get().getResolvedSchema)
}
{code}
The result will be
{code:sql}
--result1
(
`id` INT NOT NULL,
`a` ROW<`np` INT NOT NULL>
)
--result2
(
`id` INT NOT NULL,
`a` ROW<`np` INT> -- changed by
org.apache.calcite.sql.SqlDataTypeSpec#fixNullability
)
{code}
You can tell from the print result that the nullability specified by the user
does not get respect. However, this is a by-design behavior for Calcite. The
community has an in-depth discussion CALCITE-2464 on the semantics of setting
nullability for structured type.
TL;DR
!image-2023-06-09-15-06-01-322.png|width=844,height=219!
This feature is introduced in Calcite 1.19.
As a result, the DDL `a` ROW<`np` INT NOT NULL> will be rewritten to `a`
ROW<`np` INT> during the SQL-to-operation conversion.
(Please check org.apache.calcite.sql.SqlDataTypeSpec#fixNullability for more
details).
As for the plan, I assume based on the schema produced by API
{code:java}
( `id` INT NOT NULL, `a` ROW<`np` INT NOT NULL> ) {code}
the optimization rules work as expected.
The filter condition after the left outer join is
{code:java}
where a.a is null or a.a.np is null {code}
and can be reduced and pushdown as
{code:java}
where a.a is null {code}
since the nullability of a.np is always false.
And RemoveUnreachableCoalesceArgumentsRule matches the following case
{code:java}
COALESCE(a.a.np, b.a.np) c1 {code}
because a.a.np is never nullable, so the invocation of coalesce is reduced.
!image-2023-06-09-15-21-13-720.png|width=892,height=575!
h4. 3. Conclusion
So in a nutshell, we should align the structured type nullability created
through API with SQL. Last but not least, we should improve the document and
add a description of the nullability of structured type, o.w. it might not be
straightforward for users to understand.
h4.
was (Author: qingyue):
It takes some time to reason out the cause, and sorry for the late update.
h4. 1. Identify the problem
The issue has been reproduced using Flink release-1.14.6 (which depends on
calcite-1.26).
h4. 2. Why SQL gets a correct plan, while API doesn't
First, the resolved schema differs. You can verify this by the following code
snippet.
{code:scala}
@Test
def testSchema(): Unit = {
// create temporary table t1
val tableDescriptor = TableDescriptor.forConnector("datagen")
.schema(Schema.newBuilder
.column("id", DataTypes.INT.notNull)
.column("a", DataTypes.ROW(DataTypes.FIELD("np",
DataTypes.INT.notNull())).nullable)
.build)
.build
tableEnv.createTemporaryTable("t1", tableDescriptor)
// create temporary table t2
tableEnv.executeSql(
s"""
|create temporary table t2 (
| id int not null,
| a row
|) with (
| 'connector' = 'datagen'
|)
|""".stripMargin)
val catalogManager =
tableEnv.asInstanceOf[StreamTableEnvironmentImpl].getCatalogManager
val result1 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t1"))
val result2 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t2"))
println(result1.get().getResolvedSchema)
println(result2.get().getResolvedSchema)
}
{code}
The result will be
{code:sql}
--result1
(
`id` INT NOT NULL,
[jira] [Comment Edited] (FLINK-31830) Coalesce on nested fields with different nullabilities will get wrong plan
[
https://issues.apache.org/jira/browse/FLINK-31830?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17730846#comment-17730846
]
Jane Chan edited comment on FLINK-31830 at 6/9/23 7:26 AM:
---
It takes some time to reason out the cause, and sorry for the late update.
h4. 1. Identify the problem
The issue has been reproduced using Flink release-1.14.6 (which depends on
calcite-1.26).
h4. 2. Why SQL gets a correct plan, while API doesn't
First, the resolved schema differs. You can verify this by the following code
snippet.
{code:scala}
@Test
def testSchema(): Unit = {
// create temporary table t1
val tableDescriptor = TableDescriptor.forConnector("datagen")
.schema(Schema.newBuilder
.column("id", DataTypes.INT.notNull)
.column("a", DataTypes.ROW(DataTypes.FIELD("np",
DataTypes.INT.notNull())).nullable)
.build)
.build
tableEnv.createTemporaryTable("t1", tableDescriptor)
// create temporary table t2
tableEnv.executeSql(
s"""
|create temporary table t2 (
| id int not null,
| a row
|) with (
| 'connector' = 'datagen'
|)
|""".stripMargin)
val catalogManager =
tableEnv.asInstanceOf[StreamTableEnvironmentImpl].getCatalogManager
val result1 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t1"))
val result2 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t2"))
println(result1.get().getResolvedSchema)
println(result2.get().getResolvedSchema)
}
{code}
The result will be
{code:sql}
--result1
(
`id` INT NOT NULL,
`a` ROW<`np` INT NOT NULL>
)
--result2
(
`id` INT NOT NULL,
`a` ROW<`np` INT> -- changed by
org.apache.calcite.sql.SqlDataTypeSpec#fixNullability
)
{code}
You can tell from the print result that the nullability specified by the user
does not get respect. However, this is a by-design behavior for Calcite. The
community has an in-depth discussion CALCITE-2464 on the semantics of setting
nullability for structured type.
TL;DR
!image-2023-06-09-15-06-01-322.png|width=844,height=219!
This feature is introduced in Calcite 1.19. As a result, the DDL `a` ROW<`np`
INT NOT NULL> will be rewritten to `a` ROW<`np` INT> during the
SQL-to-operation conversion. (Please check
org.apache.calcite.sql.SqlDataTypeSpec#fixNullability for more details).
As for the plan, I assume based on the schema produced by API
{code:java}
( `id` INT NOT NULL, `a` ROW<`np` INT NOT NULL> ) {code}
the optimization rules work as expected.
The filter condition after the left outer join is
{code:java}
where a.a is null or a.a.np is null {code}
and can be reduced and pushdown as
{code:java}
where a.a is null {code}
since the nullability of a.np is always false.
And RemoveUnreachableCoalesceArgumentsRule matches the following case
{code:java}
COALESCE(a.a.np, b.a.np) c1 {code}
because a.a.np is never nullability, so the invocation of coalesce is reduced.
!image-2023-06-09-15-21-13-720.png|width=892,height=575!
h4. 3. Conclusion
So in a nutshell, we should align the structured type nullability created
through API with SQL. Last but not least, we should improve the document and
add a description of the nullability of structured type, o.w. it might not be
straightforward for users to understand.
h4.
was (Author: qingyue):
It takes some time to reason out the cause, and sorry for the late update.
h4. 1. Identify the problem
The issue has been reproduced using Flink release-1.14.6 (which depends on
calcite-1.26).
h4. 2. Why SQL gets a correct plan, while API doesn't
First, the resolved schema differs. You can verify this by the following code
snippet.
{code:scala}
@Test
def testSchema(): Unit = {
// create temporary table t1
val tableDescriptor = TableDescriptor.forConnector("datagen")
.schema(Schema.newBuilder
.column("id", DataTypes.INT.notNull)
.column("a", DataTypes.ROW(DataTypes.FIELD("np",
DataTypes.INT.notNull())).nullable)
.build)
.build
tableEnv.createTemporaryTable("t1", tableDescriptor)
// create temporary table t2
tableEnv.executeSql(
s"""
|create temporary table t2 (
| id int not null,
| a row
|) with (
| 'connector' = 'datagen'
|)
|""".stripMargin)
val catalogManager =
tableEnv.asInstanceOf[StreamTableEnvironmentImpl].getCatalogManager
val result1 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t1"))
val result2 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t2"))
println(result1.get().getResolvedSchema)
println(result2.get().getResolvedSchema)
}
{code}
The result will be
{code:sql}
--result1
(
`id` INT NOT NULL,
[jira] [Commented] (FLINK-31830) Coalesce on nested fields with different nullabilities will get wrong plan
[
https://issues.apache.org/jira/browse/FLINK-31830?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17730846#comment-17730846
]
Jane Chan commented on FLINK-31830:
---
It takes some time to reason out the cause, and sorry for the late update.
h4. 1. Identify the problem
The issue has been reproduced using Flink release-1.14.6 (which depends on
calcite-1.26).
h4. 2. Why SQL gets a correct plan, while API doesn't
First, the resolved schema differs. You can verify this by the following code
snippet.
{code:scala}
@Test
def testSchema(): Unit = {
// create temporary table t1
val tableDescriptor = TableDescriptor.forConnector("datagen")
.schema(Schema.newBuilder
.column("id", DataTypes.INT.notNull)
.column("a", DataTypes.ROW(DataTypes.FIELD("np",
DataTypes.INT.notNull())).nullable)
.build)
.build
tableEnv.createTemporaryTable("t1", tableDescriptor)
// create temporary table t2
tableEnv.executeSql(
s"""
|create temporary table t2 (
| id int not null,
| a row
|) with (
| 'connector' = 'datagen'
|)
|""".stripMargin)
val catalogManager =
tableEnv.asInstanceOf[StreamTableEnvironmentImpl].getCatalogManager
val result1 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t1"))
val result2 =
catalogManager.getTable(ObjectIdentifier.of("default_catalog",
"default_database", "t2"))
println(result1.get().getResolvedSchema)
println(result2.get().getResolvedSchema)
}
{code}
The result will be
{code:sql}
--result1
(
`id` INT NOT NULL,
`a` ROW<`np` INT NOT NULL>
)
--result2
(
`id` INT NOT NULL,
`a` ROW<`np` INT> -- changed by
org.apache.calcite.sql.SqlDataTypeSpec#fixNullability
)
{code}
You can tell from the print result that the nullability specified by the user
does not get respect. However, this is a by-design behavior for Calcite. The
community has an in-depth discussion CALCITE-2464 on the semantics of setting
nullability for structured type.
TL;DR
!image-2023-06-09-15-06-01-322.png|width=844,height=219!
This feature is introduced in Calcite 1.19. As a result, the DDL `a` ROW<`np`
INT NOT NULL> will be rewritten to `a` ROW<`np` INT> during the
SQL-to-operation conversion. (Please check
org.apache.calcite.sql.SqlDataTypeSpec#fixNullability for more details).
As for the plan, I assume based on the schema produced by API, the optimization
rules work as expected.
( `id` INT NOT NULL, `a` ROW<`np` INT NOT NULL> )
The filter condition after join
where a.a is null or a.a.np is null
can be reduced and pushdown as
where a.a is null
since the nullability of np is false.
And RemoveUnreachableCoalesceArgumentsRule matches the following case
COALESCE(a.a.np, b.a.np) c1
because a.a.np is never nullability, so the invocation of coalesce is reduced.
!image-2023-06-09-15-21-13-720.png|width=892,height=575!
h4. 3. Conclusion
So in a nutshell, we should align the structured type nullability created
through API with SQL. Last but not least, we should improve the document and
add a description of the nullability of structured type, o.w. it might not be
straightforward for users to understand.
h4.
> Coalesce on nested fields with different nullabilities will get wrong plan
> --
>
> Key: FLINK-31830
> URL: https://issues.apache.org/jira/browse/FLINK-31830
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.14.6
>Reporter: lincoln lee
>Assignee: Jane Chan
>Priority: Major
> Attachments: image-2023-06-09-15-06-01-322.png,
> image-2023-06-09-15-21-13-720.png
>
>
> A test case similar to FLINK-31829, only changes the nullable field `a.np` to
> not null, will get a wrong plan in 1.14.x (reported from the community user):
> {code}
> @Test
> def testCoalesceOnNestedColumns(): Unit = {
> val tEnv = util.tableEnv
> val tableDescriptor = TableDescriptor.forConnector("datagen")
> .schema(Schema.newBuilder
> .column("id", DataTypes.INT.notNull)
> .column("a", DataTypes.ROW(DataTypes.FIELD("np",
> DataTypes.INT.notNull())).nullable)
> .build)
> .build
> tEnv.createTemporaryTable("t1", tableDescriptor)
> tEnv.createTemporaryTable("t2", tableDescriptor)
> val res = tEnv.executeSql("EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np)
> c1, IFNULL(a.a.np, b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a
> is null or a.a.np is null")
> res.print()
> }
> == Abstract Syntax Tree ==
> LogicalProject(id=[$0], c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])
> +- LogicalFilter(condition=[OR(IS NULL($1), IS NULL(CAST($1.np):INTEGER))])
>+-
[jira] [Updated] (FLINK-31830) Coalesce on nested fields with different nullabilities will get wrong plan
[
https://issues.apache.org/jira/browse/FLINK-31830?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31830:
--
Attachment: image-2023-06-09-15-21-13-720.png
> Coalesce on nested fields with different nullabilities will get wrong plan
> --
>
> Key: FLINK-31830
> URL: https://issues.apache.org/jira/browse/FLINK-31830
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.14.6
>Reporter: lincoln lee
>Assignee: Jane Chan
>Priority: Major
> Attachments: image-2023-06-09-15-06-01-322.png,
> image-2023-06-09-15-21-13-720.png
>
>
> A test case similar to FLINK-31829, only changes the nullable field `a.np` to
> not null, will get a wrong plan in 1.14.x (reported from the community user):
> {code}
> @Test
> def testCoalesceOnNestedColumns(): Unit = {
> val tEnv = util.tableEnv
> val tableDescriptor = TableDescriptor.forConnector("datagen")
> .schema(Schema.newBuilder
> .column("id", DataTypes.INT.notNull)
> .column("a", DataTypes.ROW(DataTypes.FIELD("np",
> DataTypes.INT.notNull())).nullable)
> .build)
> .build
> tEnv.createTemporaryTable("t1", tableDescriptor)
> tEnv.createTemporaryTable("t2", tableDescriptor)
> val res = tEnv.executeSql("EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np)
> c1, IFNULL(a.a.np, b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a
> is null or a.a.np is null")
> res.print()
> }
> == Abstract Syntax Tree ==
> LogicalProject(id=[$0], c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])
> +- LogicalFilter(condition=[OR(IS NULL($1), IS NULL(CAST($1.np):INTEGER))])
>+- LogicalJoin(condition=[=($0, $2)], joinType=[left])
> :- LogicalTableScan(table=[[default_catalog, default_database, t1]])
> +- LogicalTableScan(table=[[default_catalog, default_database, t2]])
> {code}
> the top project in the ast is wrong: `LogicalProject(id=[$0],
> c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])`, the
> `c1=[CAST($1.np):INTEGER]` relate to `COALESCE(a.a.np, b.a.np) c1` is
> incorrect,
> but this works fine when using sql ddl to create tables
> {code}
> @Test
> def testCoalesceOnNestedColumns2(): Unit = {
> val tEnv = util.tableEnv
> tEnv.executeSql(
> s"""
> |create temporary table t1 (
> | id int not null,
> | a row
> |) with (
> | 'connector' = 'datagen'
> |)
> |""".stripMargin)
> tEnv.executeSql(
> s"""
> |create temporary table t2 (
> | id int not null,
> | a row
> |) with (
> | 'connector' = 'datagen'
> |)
> |""".stripMargin)
> val res = tEnv.executeSql(
> "EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np) c1, IFNULL(a.a.np,
> b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a is null or a.a.np
> is null")
> res.print()
> }
> {code}
> from 1.15, the coalesce will be a new builtin function, and the ast looks
> correct in version 1.15+, while before 1.15 it was rewritten as `case when`
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31830) Coalesce on nested fields with different nullabilities will get wrong plan
[
https://issues.apache.org/jira/browse/FLINK-31830?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31830:
--
Attachment: image-2023-06-09-15-06-01-322.png
> Coalesce on nested fields with different nullabilities will get wrong plan
> --
>
> Key: FLINK-31830
> URL: https://issues.apache.org/jira/browse/FLINK-31830
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.14.6
>Reporter: lincoln lee
>Assignee: Jane Chan
>Priority: Major
> Attachments: image-2023-06-09-15-06-01-322.png
>
>
> A test case similar to FLINK-31829, only changes the nullable field `a.np` to
> not null, will get a wrong plan in 1.14.x (reported from the community user):
> {code}
> @Test
> def testCoalesceOnNestedColumns(): Unit = {
> val tEnv = util.tableEnv
> val tableDescriptor = TableDescriptor.forConnector("datagen")
> .schema(Schema.newBuilder
> .column("id", DataTypes.INT.notNull)
> .column("a", DataTypes.ROW(DataTypes.FIELD("np",
> DataTypes.INT.notNull())).nullable)
> .build)
> .build
> tEnv.createTemporaryTable("t1", tableDescriptor)
> tEnv.createTemporaryTable("t2", tableDescriptor)
> val res = tEnv.executeSql("EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np)
> c1, IFNULL(a.a.np, b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a
> is null or a.a.np is null")
> res.print()
> }
> == Abstract Syntax Tree ==
> LogicalProject(id=[$0], c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])
> +- LogicalFilter(condition=[OR(IS NULL($1), IS NULL(CAST($1.np):INTEGER))])
>+- LogicalJoin(condition=[=($0, $2)], joinType=[left])
> :- LogicalTableScan(table=[[default_catalog, default_database, t1]])
> +- LogicalTableScan(table=[[default_catalog, default_database, t2]])
> {code}
> the top project in the ast is wrong: `LogicalProject(id=[$0],
> c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])`, the
> `c1=[CAST($1.np):INTEGER]` relate to `COALESCE(a.a.np, b.a.np) c1` is
> incorrect,
> but this works fine when using sql ddl to create tables
> {code}
> @Test
> def testCoalesceOnNestedColumns2(): Unit = {
> val tEnv = util.tableEnv
> tEnv.executeSql(
> s"""
> |create temporary table t1 (
> | id int not null,
> | a row
> |) with (
> | 'connector' = 'datagen'
> |)
> |""".stripMargin)
> tEnv.executeSql(
> s"""
> |create temporary table t2 (
> | id int not null,
> | a row
> |) with (
> | 'connector' = 'datagen'
> |)
> |""".stripMargin)
> val res = tEnv.executeSql(
> "EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np) c1, IFNULL(a.a.np,
> b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a is null or a.a.np
> is null")
> res.print()
> }
> {code}
> from 1.15, the coalesce will be a new builtin function, and the ast looks
> correct in version 1.15+, while before 1.15 it was rewritten as `case when`
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-32219) SQL client hangs when executing EXECUTE PLAN
[
https://issues.apache.org/jira/browse/FLINK-32219?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17729133#comment-17729133
]
Jane Chan commented on FLINK-32219:
---
Hi, [~xu_shuai_] Thanks for reporting this issue.
OperationExecutor failed to consider the types of
CompileAndExecutePlanOperation and ExecutePlanOperation and instead used
callOperation by default, resulting in no job ID being returned. I'd like to
fix this issue. cc [~shengkai]
> SQL client hangs when executing EXECUTE PLAN
>
>
> Key: FLINK-32219
> URL: https://issues.apache.org/jira/browse/FLINK-32219
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Client
>Affects Versions: 1.17.1
>Reporter: Shuai Xu
>Priority: Major
>
> I compiled a plan for an INSERT statement and executed the plan, but the SQL
> client became unresponsive when executing the EXECUTE PLAN statement. I
> confirmed that the Flink job is running normally by checking the Flink
> dashboard. The only issue is that the SQL client becomes stuck and cannot
> accept new commands. I printed the stack trace of the SQL client process, and
> here is a part of it for reference.
> {code:java}
> "pool-2-thread-1" #30 prio=5 os_prio=31 tid=0x0001172e5000 nid=0x6d03
> waiting on condition [0x000173e01000]
>java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for <0x00076e72af20> (a
> java.util.concurrent.CompletableFuture$Signaller)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at
> java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1707)
> at
> java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3323)
> at
> java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1742)
> at
> java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
> at
> org.apache.flink.table.api.internal.InsertResultProvider.hasNext(InsertResultProvider.java:83)
> at
> org.apache.flink.table.api.internal.InsertResultProvider.access$200(InsertResultProvider.java:37)
> at
> org.apache.flink.table.api.internal.InsertResultProvider$Iterator.hasNext(InsertResultProvider.java:106)
> at java.util.Iterator.forEachRemaining(Iterator.java:115)
> at
> org.apache.flink.util.CollectionUtil.iteratorToList(CollectionUtil.java:115)
> at
> org.apache.flink.table.gateway.service.result.ResultFetcher.fromTableResult(ResultFetcher.java:163)
> at
> org.apache.flink.table.gateway.service.operation.OperationExecutor.callOperation(OperationExecutor.java:542)
> at
> org.apache.flink.table.gateway.service.operation.OperationExecutor.executeOperation(OperationExecutor.java:440)
> at
> org.apache.flink.table.gateway.service.operation.OperationExecutor.executeStatement(OperationExecutor.java:195)
> at
> org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.lambda$executeStatement$1(SqlGatewayServiceImpl.java:212)
> at
> org.apache.flink.table.gateway.service.SqlGatewayServiceImpl$$Lambda$389/1391083077.apply(Unknown
> Source)
> at
> org.apache.flink.table.gateway.service.operation.OperationManager.lambda$submitOperation$1(OperationManager.java:119)
> at
> org.apache.flink.table.gateway.service.operation.OperationManager$$Lambda$390/208625838.call(Unknown
> Source)
> at
> org.apache.flink.table.gateway.service.operation.OperationManager$Operation.lambda$run$0(OperationManager.java:258)
> at
> org.apache.flink.table.gateway.service.operation.OperationManager$Operation$$Lambda$392/670621032.run(Unknown
> Source)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:750)
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-32245) NonDeterministicTests #testTemporalFunctionsInBatchMode failure masked due to incorrect test initialization
[ https://issues.apache.org/jira/browse/FLINK-32245?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-32245: -- Description: After introducing FLINK-30006, the test case NonDeterministicTests #testTemporalFunctionsInBatchMode has been consistently failing due to incorrect test initialization. However, this failure has been masked because the test class name ends with "Tests", causing the CI to skip the test case, which has been further validated by searching through the historical logs of the CI. This issue needs to be addressed, and the test case should be executed to ensure proper testing. > NonDeterministicTests #testTemporalFunctionsInBatchMode failure masked due to > incorrect test initialization > --- > > Key: FLINK-32245 > URL: https://issues.apache.org/jira/browse/FLINK-32245 > Project: Flink > Issue Type: Bug > Components: Table SQL / Planner >Affects Versions: 1.18.0, 1.17.1 >Reporter: Jane Chan >Priority: Major > Fix For: 1.18.0 > > > After introducing FLINK-30006, the test case NonDeterministicTests > #testTemporalFunctionsInBatchMode has been consistently failing due to > incorrect test initialization. > > However, this failure has been masked because the test class name ends with > "Tests", causing the CI to skip the test case, which has been further > validated by searching through the historical logs of the CI. > This issue needs to be addressed, and the test case should be executed to > ensure proper testing. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-32245) NonDeterministicTests #testTemporalFunctionsInBatchMode failure masked due to incorrect test initialization
[ https://issues.apache.org/jira/browse/FLINK-32245?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-32245: -- Description: The test case NonDeterministicTests #testTemporalFunctionsInBatchMode has been consistently failing due to incorrect test initialization. However, this failure has been masked because the test class name ends with "Tests", causing the CI to skip the test case, which has been further validated by searching through the historical logs of the CI. This issue needs to be addressed, and the test case should be executed to ensure proper testing. was: After introducing FLINK-30006, the test case NonDeterministicTests #testTemporalFunctionsInBatchMode has been consistently failing due to incorrect test initialization. However, this failure has been masked because the test class name ends with "Tests", causing the CI to skip the test case, which has been further validated by searching through the historical logs of the CI. This issue needs to be addressed, and the test case should be executed to ensure proper testing. > NonDeterministicTests #testTemporalFunctionsInBatchMode failure masked due to > incorrect test initialization > --- > > Key: FLINK-32245 > URL: https://issues.apache.org/jira/browse/FLINK-32245 > Project: Flink > Issue Type: Bug > Components: Table SQL / Planner >Affects Versions: 1.18.0, 1.17.1 >Reporter: Jane Chan >Priority: Major > Fix For: 1.18.0 > > > The test case NonDeterministicTests #testTemporalFunctionsInBatchMode has > been consistently failing due to incorrect test initialization. > > However, this failure has been masked because the test class name ends with > "Tests", causing the CI to skip the test case, which has been further > validated by searching through the historical logs of the CI. > This issue needs to be addressed, and the test case should be executed to > ensure proper testing. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-32245) NonDeterministicTests #testTemporalFunctionsInBatchMode failure masked due to incorrect test initialization
Jane Chan created FLINK-32245: - Summary: NonDeterministicTests #testTemporalFunctionsInBatchMode failure masked due to incorrect test initialization Key: FLINK-32245 URL: https://issues.apache.org/jira/browse/FLINK-32245 Project: Flink Issue Type: Bug Components: Table SQL / Planner Affects Versions: 1.17.1, 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-32053) Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan
[
https://issues.apache.org/jira/browse/FLINK-32053?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-32053:
--
Labels: pull-request-available (was: )
> Introduce StateMetadata to ExecNode to support configure operator-level state
> TTL via CompiledPlan
> --
>
> Key: FLINK-32053
> URL: https://issues.apache.org/jira/browse/FLINK-32053
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Assignee: Jane Chan
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.18.0
>
>
> This subtask should introduce StateMetadata to all ExecNodes that translate
> to stateful operators, changing the way how `#translateToPlanInternal` get
> the state retention time. The affected `ExecNode` list
> {code:java}
> StreamExecChangelogNormalize
> StreamExecDeduplicate
> StreamExecGlobalGroupAggregate
> StreamExecGroupAggregate
> StreamExecIncrementalGroupAggregate
> StreamExecJoin
> StreamExecLimit
> StreamExecLookupJoin
> StreamExecRank
> StreamExecSink
> StreamExecSortLimit
> {code}
> Since we have upgraded some `ExecNode`s to version 2, we have to test the
> following 3 parts:
> 1. the plans serialized using version 1 can be deserialized using the current
> version.
> 2. the plans with the current version SerDe work as expected.
> 3. The way by modifying the JSON content to change state TTL works as
> expected.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] (FLINK-32053) Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan
[ https://issues.apache.org/jira/browse/FLINK-32053 ]
Jane Chan deleted comment on FLINK-32053:
---
was (Author: qingyue):
Wait for FLINK-32064 to be merged
> Introduce StateMetadata to ExecNode to support configure operator-level state
> TTL via CompiledPlan
> --
>
> Key: FLINK-32053
> URL: https://issues.apache.org/jira/browse/FLINK-32053
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Assignee: Jane Chan
>Priority: Major
> Fix For: 1.18.0
>
>
> This subtask should introduce StateMetadata to all ExecNodes that translate
> to stateful operators, changing the way how `#translateToPlanInternal` get
> the state retention time. The affected `ExecNode` list
> {code:java}
> StreamExecChangelogNormalize
> StreamExecDeduplicate
> StreamExecGlobalGroupAggregate
> StreamExecGroupAggregate
> StreamExecIncrementalGroupAggregate
> StreamExecJoin
> StreamExecLimit
> StreamExecLookupJoin
> StreamExecRank
> StreamExecSink
> StreamExecSortLimit
> {code}
> Since we have upgraded some `ExecNode`s to version 2, we have to test the
> following 3 parts:
> 1. the plans serialized using version 1 can be deserialized using the current
> version.
> 2. the plans with the current version SerDe work as expected.
> 3. The way by modifying the JSON content to change state TTL works as
> expected.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Closed] (FLINK-31917) Loss of Idempotence in JsonSerDe Round Trip for AggregateCall and RexNode
[
https://issues.apache.org/jira/browse/FLINK-31917?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan closed FLINK-31917.
-
Resolution: Fixed
Merged into master: 333e023196d90d265c286632f8c01c41b8911ef8
> Loss of Idempotence in JsonSerDe Round Trip for AggregateCall and RexNode
> -
>
> Key: FLINK-31917
> URL: https://issues.apache.org/jira/browse/FLINK-31917
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Assignee: Jane Chan
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.18.0
>
>
> JsonSerDeTestUtil#testJsonRoundTrip only checks the equality between spec and
> deserialized object. Some corner cases are detected when serializing the
> deserialized object again.
> {code:java}
> static T testJsonRoundTrip(SerdeContext serdeContext, T spec, Class
> clazz)
> throws IOException {
> String actualJson = toJson(serdeContext, spec);
> T actual = toObject(serdeContext, actualJson, clazz);
> assertThat(actual).isEqualTo(spec);
> assertThat(actualJson).isEqualTo(toJson(serdeContext, actual)); //
> this will eval some corner cases
> return actual;
> }
> {code}
> The discovered corner cases are listed as follows.
> h5. 1. SerDe for AggregateCall
> When deserializing the aggregate call, we should check the JsonNodeType to
> avoid converting null to "null" string.
> [https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AggregateCallJsonDeserializer.java#L64]
> h5. Suggested Fix
> {code:java}
> JsonNode nameNode = jsonNode.required(FIELD_NAME_NAME);
> final String name = JsonNodeType.NULL ? null : nameNode.asText();
> {code}
> h5. 2. SerDe for RexNode
> RexNodeJsonSerdeTest#testSystemFunction should create the temporary system
> function instead of the temporary catalog function.
> [https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerdeTest.java#L209]
> h5. Suggested Fix
> Use functionCatalog#registerTemporarySystemFunction to test.
> h5. 3. About RexLiteral type
> RexNodeJsonSerdeTest#testRexNodeSerde has a test spec as follows
> {code:java}
> //This will create the literal with DOUBLE as the literal type, and DECIMAL
> as the broad type of this literal. You can refer to Calcite for more details
> rexBuilder.makeExactLiteral(BigDecimal.valueOf(Double.MAX_VALUE),
> FACTORY.createSqlType(SqlTypeName.DOUBLE))
> {code}
> The RexNodeJsonSerializer uses `typeName`(which is DECIMAL) as the literal's
> type, as a result, the rel data type is serialized as double, but the value
> is serialized as a string (in case lost the precision)
> [https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerializer.java#L197]
> And then, during the deserialization, according to the JSON, the deserialized
> literal will assign DOUBLE as the literal type and the broad type of the
> literal.
> This will cause the comparison failure
> {code:java}
> expected: {"kind": "LITERAL", "value": "1.7976931348623157E+308"}
> actual: {"kind": "LITERAL", "value": 1.7976931348623157E+308}
> {code}
> h5. Suggested Fix
> SARG is a special case and can be coped first, and for the rest type, we can
> use literal.getType().getSqlTypeName() instead of literal.getTypeName().
> {code:java}
> // first cope with SARG type
> if (literal.getTypeName() == SARG) {
> serializeSargValue(
> (Sarg) value, literal.getType().getSqlTypeName(), gen,
> serializerProvider);
> } else {
> serializeLiteralValue(
> value,
> literal.getType().getSqlTypeName(),
> gen,
> serializerProvider);
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Closed] (FLINK-31884) Upgrade ExecNode to new version causes the old serialized plan failed to pass Json SerDe round trip
[
https://issues.apache.org/jira/browse/FLINK-31884?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan closed FLINK-31884.
-
Resolution: Fixed
Fixed in master: 3664609c7622ccae80e36e85099a1b79b5935fe9
> Upgrade ExecNode to new version causes the old serialized plan failed to pass
> Json SerDe round trip
> ---
>
> Key: FLINK-31884
> URL: https://issues.apache.org/jira/browse/FLINK-31884
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Assignee: Jane Chan
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.18.0
>
>
> h4. How to Reproduce
> Firstly, add a test to dump the compiled plan JSON.
> {code:java}
> @Test
> public void debug() {
> tableEnv.executeSql("create table foo (f0 int, f1 string) with
> ('connector' = 'datagen')");
> tableEnv.executeSql("create table bar (f0 int, f1 string) with
> ('connector' = 'print')");
> tableEnv.compilePlanSql("insert into bar select * from foo")
> .writeToFile(new File("/path/to/debug.json"));
> }
> {code}
> The JSON context is as follows
> {code:json}
> {
> "flinkVersion" : "1.18",
> "nodes" : [ {
> "id" : 1,
> "type" : "stream-exec-table-source-scan_1",
> "scanTableSource" : {
> "table" : {
> "identifier" : "`default_catalog`.`default_database`.`foo`",
> "resolvedTable" : {
> "schema" : {
> "columns" : [ {
> "name" : "f0",
> "dataType" : "INT"
> }, {
> "name" : "f1",
> "dataType" : "VARCHAR(2147483647)"
> } ],
> "watermarkSpecs" : [ ]
> },
> "partitionKeys" : [ ],
> "options" : {
> "connector" : "datagen"
> }
> }
> }
> },
> "outputType" : "ROW<`f0` INT, `f1` VARCHAR(2147483647)>",
> "description" : "TableSourceScan(table=[[default_catalog,
> default_database, foo]], fields=[f0, f1])",
> "inputProperties" : [ ]
> }, {
> "id" : 2,
> "type" : "stream-exec-sink_1",
> "configuration" : {
> "table.exec.sink.keyed-shuffle" : "AUTO",
> "table.exec.sink.not-null-enforcer" : "ERROR",
> "table.exec.sink.type-length-enforcer" : "IGNORE",
> "table.exec.sink.upsert-materialize" : "AUTO"
> },
> "dynamicTableSink" : {
> "table" : {
> "identifier" : "`default_catalog`.`default_database`.`bar`",
> "resolvedTable" : {
> "schema" : {
> "columns" : [ {
> "name" : "f0",
> "dataType" : "INT"
> }, {
> "name" : "f1",
> "dataType" : "VARCHAR(2147483647)"
> } ],
> "watermarkSpecs" : [ ]
> },
> "partitionKeys" : [ ],
> "options" : {
> "connector" : "print"
> }
> }
> }
> },
> "inputChangelogMode" : [ "INSERT" ],
> "inputProperties" : [ {
> "requiredDistribution" : {
> "type" : "UNKNOWN"
> },
> "damBehavior" : "PIPELINED",
> "priority" : 0
> } ],
> "outputType" : "ROW<`f0` INT, `f1` VARCHAR(2147483647)>",
> "description" : "Sink(table=[default_catalog.default_database.bar],
> fields=[f0, f1])"
> } ],
> "edges" : [ {
> "source" : 1,
> "target" : 2,
> "shuffle" : {
> "type" : "FORWARD"
> },
> "shuffleMode" : "PIPELINED"
> } ]
> }
> {code}
> Then upgrade the StreamExecSink to a new version
> {code:java}
> @ExecNodeMetadata(
> name = "stream-exec-sink",
> version = 1,
> consumedOptions = {
> "table.exec.sink.not-null-enforcer",
> "table.exec.sink.type-length-enforcer",
> "table.exec.sink.upsert-materialize",
> "table.exec.sink.keyed-shuffle"
> },
> producedTransformations = {
> CommonExecSink.CONSTRAINT_VALIDATOR_TRANSFORMATION,
> CommonExecSink.PARTITIONER_TRANSFORMATION,
> CommonExecSink.UPSERT_MATERIALIZE_TRANSFORMATION,
> CommonExecSink.TIMESTAMP_INSERTER_TRANSFORMATION,
> CommonExecSink.SINK_TRANSFORMATION
> },
> minPlanVersion = FlinkVersion.v1_15,
> minStateVersion = FlinkVersion.v1_15)
> @ExecNodeMetadata(
> name = "stream-exec-sink",
> version = 2,
> consumedOptions = {
> "table.exec.sink.not-null-enforcer",
> "table.exec.sink.type-length-enforcer",
> "table.exec.sink.upsert-materialize",
> "table.exec.sink.keyed-shuffle"
> },
> producedTransformations = {
>
[jira] [Commented] (FLINK-32064) Add subdirectory of test output file for JsonPlanTest to indicate the plan's version
[ https://issues.apache.org/jira/browse/FLINK-32064?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17722768#comment-17722768 ] Jane Chan commented on FLINK-32064: --- After an offline discussion with [~godfreyhe], we have decided to select a portion of the existing JSON files as test specs for the upgrade tests, and directly adapt the remaining tests to the new exec version to avoid excessive redundancy in the test files. > Add subdirectory of test output file for JsonPlanTest to indicate the plan's > version > - > > Key: FLINK-32064 > URL: https://issues.apache.org/jira/browse/FLINK-32064 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > This is a preparation for upgrading some ExecNodes(which translate to > stateful operators) version to 2 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-32064) Add subdirectory of test output file for JsonPlanTest to indicate the plan's version
[ https://issues.apache.org/jira/browse/FLINK-32064?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-32064: -- Release Note: (was: After an offline discussion with Godfrey, we have decided to select a portion of the existing JSON files as test specs for the upgrade tests, and directly adapt the remaining tests to the new exec version to avoid excessive redundancy in the test files.) > Add subdirectory of test output file for JsonPlanTest to indicate the plan's > version > - > > Key: FLINK-32064 > URL: https://issues.apache.org/jira/browse/FLINK-32064 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > This is a preparation for upgrading some ExecNodes(which translate to > stateful operators) version to 2 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Closed] (FLINK-32064) Add subdirectory of test output file for JsonPlanTest to indicate the plan's version
[ https://issues.apache.org/jira/browse/FLINK-32064?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan closed FLINK-32064. - Release Note: After an offline discussion with Godfrey, we have decided to select a portion of the existing JSON files as test specs for the upgrade tests, and directly adapt the remaining tests to the new exec version to avoid excessive redundancy in the test files. Resolution: Won't Do > Add subdirectory of test output file for JsonPlanTest to indicate the plan's > version > - > > Key: FLINK-32064 > URL: https://issues.apache.org/jira/browse/FLINK-32064 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > This is a preparation for upgrading some ExecNodes(which translate to > stateful operators) version to 2 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31950) Introduce StateMetadata
[ https://issues.apache.org/jira/browse/FLINK-31950?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31950: -- Summary: Introduce StateMetadata (was: Introduce StateMetadata and StateMetadataJson SerDe) > Introduce StateMetadata > --- > > Key: FLINK-31950 > URL: https://issues.apache.org/jira/browse/FLINK-31950 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > According to the FLIP design, we're about to introduce > * StateMetadata, which describes the TTL attribute of the stateful stream > operator. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31950) Introduce StateMetadata and StateMetadataJson SerDe
[ https://issues.apache.org/jira/browse/FLINK-31950?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31950: -- Description: According to the FLIP design, we're about to introduce * StateMetadata, which describes the TTL attribute of the stateful stream operator. was: According to the FLIP design, we're about to introduce * StateMetadata, which describes the TTL attribute of the stateful stream operator. * StateMetadata SerDerializers. > Introduce StateMetadata and StateMetadataJson SerDe > --- > > Key: FLINK-31950 > URL: https://issues.apache.org/jira/browse/FLINK-31950 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > According to the FLIP design, we're about to introduce > * StateMetadata, which describes the TTL attribute of the stateful stream > operator. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-32053) Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan
[
https://issues.apache.org/jira/browse/FLINK-32053?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17721994#comment-17721994
]
Jane Chan commented on FLINK-32053:
---
Wait for FLINK-32064 to be merged
> Introduce StateMetadata to ExecNode to support configure operator-level state
> TTL via CompiledPlan
> --
>
> Key: FLINK-32053
> URL: https://issues.apache.org/jira/browse/FLINK-32053
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Assignee: Jane Chan
>Priority: Major
> Fix For: 1.18.0
>
>
> This subtask should introduce StateMetadata to all ExecNodes that translate
> to stateful operators, changing the way how `#translateToPlanInternal` get
> the state retention time. The affected `ExecNode` list
> {code:java}
> StreamExecChangelogNormalize
> StreamExecDeduplicate
> StreamExecGlobalGroupAggregate
> StreamExecGroupAggregate
> StreamExecIncrementalGroupAggregate
> StreamExecJoin
> StreamExecLimit
> StreamExecLookupJoin
> StreamExecRank
> StreamExecSink
> StreamExecSortLimit
> {code}
> Since we have upgraded some `ExecNode`s to version 2, we have to test the
> following 3 parts:
> 1. the plans serialized using version 1 can be deserialized using the current
> version.
> 2. the plans with the current version SerDe work as expected.
> 3. The way by modifying the JSON content to change state TTL works as
> expected.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-32064) Add subdirectory of test output file for JsonPlanTest to indicate the plan's version
[ https://issues.apache.org/jira/browse/FLINK-32064?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-32064: -- Summary: Add subdirectory of test output file for JsonPlanTest to indicate the plan's version (was: Add sub-directory of test output file for JsonPlanTest to indicate the plan's version) > Add subdirectory of test output file for JsonPlanTest to indicate the plan's > version > - > > Key: FLINK-32064 > URL: https://issues.apache.org/jira/browse/FLINK-32064 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Priority: Major > Fix For: 1.18.0 > > > This is a preparation for upgrading some ExecNodes(which translate to > stateful operators) version to 2 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-32064) Add sub-directory of test output file for JsonPlanTest to indicate the plan's version
[ https://issues.apache.org/jira/browse/FLINK-32064?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-32064: -- Description: This is a preparation for upgrading some ExecNodes(which translate to stateful operators) version to 2 (was: This is a preparation for upgrade ExecNode version to 2) > Add sub-directory of test output file for JsonPlanTest to indicate the > plan's version > -- > > Key: FLINK-32064 > URL: https://issues.apache.org/jira/browse/FLINK-32064 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Priority: Major > Fix For: 1.18.0 > > > This is a preparation for upgrading some ExecNodes(which translate to > stateful operators) version to 2 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-32064) Add sub-directory of test output file for JsonPlanTest to indicate the plan's version
[ https://issues.apache.org/jira/browse/FLINK-32064?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-32064: -- Description: This is a preparation for upgrade ExecNode version to 2 > Add sub-directory of test output file for JsonPlanTest to indicate the > plan's version > -- > > Key: FLINK-32064 > URL: https://issues.apache.org/jira/browse/FLINK-32064 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Priority: Major > Fix For: 1.18.0 > > > This is a preparation for upgrade ExecNode version to 2 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-32064) Add sub-directory of test output file for JsonPlanTest to indicate the plan's version
Jane Chan created FLINK-32064: - Summary: Add sub-directory of test output file for JsonPlanTest to indicate the plan's version Key: FLINK-32064 URL: https://issues.apache.org/jira/browse/FLINK-32064 Project: Flink Issue Type: Sub-task Components: Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-32053) Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan
[
https://issues.apache.org/jira/browse/FLINK-32053?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-32053:
--
Description:
This subtask should introduce StateMetadata to all ExecNodes that translate to
stateful operators, changing the way how `#translateToPlanInternal` get the
state retention time. The affected `ExecNode` list
{code:java}
StreamExecChangelogNormalize
StreamExecDeduplicate
StreamExecGlobalGroupAggregate
StreamExecGroupAggregate
StreamExecIncrementalGroupAggregate
StreamExecJoin
StreamExecLimit
StreamExecLookupJoin
StreamExecRank
StreamExecSink
StreamExecSortLimit
{code}
Since we have upgraded some `ExecNode`s to version 2, we have to test the
following 3 parts:
1. the plans serialized using version 1 can be deserialized using the current
version.
2. the plans with the current version SerDe work as expected.
3. The way by modifying the JSON content to change state TTL works as expected.
> Introduce StateMetadata to ExecNode to support configure operator-level state
> TTL via CompiledPlan
> --
>
> Key: FLINK-32053
> URL: https://issues.apache.org/jira/browse/FLINK-32053
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: 1.18.0
>
>
> This subtask should introduce StateMetadata to all ExecNodes that translate
> to stateful operators, changing the way how `#translateToPlanInternal` get
> the state retention time. The affected `ExecNode` list
> {code:java}
> StreamExecChangelogNormalize
> StreamExecDeduplicate
> StreamExecGlobalGroupAggregate
> StreamExecGroupAggregate
> StreamExecIncrementalGroupAggregate
> StreamExecJoin
> StreamExecLimit
> StreamExecLookupJoin
> StreamExecRank
> StreamExecSink
> StreamExecSortLimit
> {code}
> Since we have upgraded some `ExecNode`s to version 2, we have to test the
> following 3 parts:
> 1. the plans serialized using version 1 can be deserialized using the current
> version.
> 2. the plans with the current version SerDe work as expected.
> 3. The way by modifying the JSON content to change state TTL works as
> expected.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-32053) Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan
Jane Chan created FLINK-32053: - Summary: Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan Key: FLINK-32053 URL: https://issues.apache.org/jira/browse/FLINK-32053 Project: Flink Issue Type: Sub-task Components: Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-32052) Introduce left and right state retention time to StreamingJoinOperator
Jane Chan created FLINK-32052: - Summary: Introduce left and right state retention time to StreamingJoinOperator Key: FLINK-32052 URL: https://issues.apache.org/jira/browse/FLINK-32052 Project: Flink Issue Type: Sub-task Components: Table SQL / Runtime Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 According to the FLIP design, we should introduce separate TTL variables to the TwoInputStreamOperator, like StreamingJoinOperator and StreamingAntiJoinOperator. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31950) Introduce StateMetadata and StateMetadataJson SerDe
[ https://issues.apache.org/jira/browse/FLINK-31950?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31950: -- Summary: Introduce StateMetadata and StateMetadataJson SerDe (was: Introduce StateMetadata and JSON SerDe) > Introduce StateMetadata and StateMetadataJson SerDe > --- > > Key: FLINK-31950 > URL: https://issues.apache.org/jira/browse/FLINK-31950 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > According to the FLIP design, we're about to introduce > * StateMetadata, which describes the TTL attribute of the stateful stream > operator. > * StateMetadata SerDerializers. > * For StreamExecNodes which translate to stateful and TTL-sensitive > operators, change #translateToPlanInternal on how to fetch the TTL > configuration during plan deserialization. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31950) Introduce StateMetadata and StateMetadataJson SerDe
[ https://issues.apache.org/jira/browse/FLINK-31950?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31950: -- Description: According to the FLIP design, we're about to introduce * StateMetadata, which describes the TTL attribute of the stateful stream operator. * StateMetadata SerDerializers. was: According to the FLIP design, we're about to introduce * StateMetadata, which describes the TTL attribute of the stateful stream operator. * StateMetadata SerDerializers. * For StreamExecNodes which translate to stateful and TTL-sensitive operators, change #translateToPlanInternal on how to fetch the TTL configuration during plan deserialization. > Introduce StateMetadata and StateMetadataJson SerDe > --- > > Key: FLINK-31950 > URL: https://issues.apache.org/jira/browse/FLINK-31950 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > According to the FLIP design, we're about to introduce > * StateMetadata, which describes the TTL attribute of the stateful stream > operator. > * StateMetadata SerDerializers. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31950) Introduce StateMetadata and JSON SerDe
[ https://issues.apache.org/jira/browse/FLINK-31950?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31950: -- Summary: Introduce StateMetadata and JSON SerDe (was: Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan) > Introduce StateMetadata and JSON SerDe > -- > > Key: FLINK-31950 > URL: https://issues.apache.org/jira/browse/FLINK-31950 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.18.0 > > > According to the FLIP design, we're about to introduce > * StateMetadata, which describes the TTL attribute of the stateful stream > operator. > * StateMetadata SerDerializers. > * For StreamExecNodes which translate to stateful and TTL-sensitive > operators, change #translateToPlanInternal on how to fetch the TTL > configuration during plan deserialization. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-31957) Add documentation for the user story
Jane Chan created FLINK-31957: - Summary: Add documentation for the user story Key: FLINK-31957 URL: https://issues.apache.org/jira/browse/FLINK-31957 Project: Flink Issue Type: Sub-task Components: Documentation, Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 Add documentation on how to use compiled plan to configure operator-level state TTL. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31956) Extend the CompiledPlan to read from/write to Flink's FileSystem
[
https://issues.apache.org/jira/browse/FLINK-31956?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31956:
--
Description:
At present, COMPILE/EXECUTE PLAN FOR '${plan.json}' only supports writing
to/reading from a local file without the scheme. We propose to extend the
support for Flink's FileSystem.
{code:sql}
-- before
COMPILE PLAN FOR '/tmp/foo/bar.json'
EXECUTE PLAN FOR '/tmp/foo/bar.json'
-- after
COMPILE PLAN FOR 'file:///tmp/foo/bar.json'
COMPILE PLAN FOR 'hdfs:///tmp/foo/bar.json'
COMPILE PLAN FOR 's3:///tmp/foo/bar.json'
COMPILE PLAN FOR 'oss:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'file:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'hdfs:///tmp/foo/bar.json'
EXECUTE PLAN FOR 's3:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'oss:///tmp/foo/bar.json' {code}
was:
At present, COMPILE/EXECUTE PLAN FOR '${plan.json}' only supports writing
to/reading from a local file without the scheme. We propose to extend the
support for Flink's FileSystem.
{code:java}
-- before
COMPILE PLAN FOR '/tmp/foo/bar.json'
EXECUTE PLAN FOR '/tmp/foo/bar.json'
-- after
COMPILE PLAN FOR 'file:///tmp/foo/bar.json'
COMPILE PLAN FOR 'hdfs:///tmp/foo/bar.json'
COMPILE PLAN FOR 's3:///tmp/foo/bar.json'
COMPILE PLAN FOR 'oss:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'file:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'hdfs:///tmp/foo/bar.json'
EXECUTE PLAN FOR 's3:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'oss:///tmp/foo/bar.json' {code}
> Extend the CompiledPlan to read from/write to Flink's FileSystem
>
>
> Key: FLINK-31956
> URL: https://issues.apache.org/jira/browse/FLINK-31956
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Client, Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: 1.18.0
>
>
> At present, COMPILE/EXECUTE PLAN FOR '${plan.json}' only supports writing
> to/reading from a local file without the scheme. We propose to extend the
> support for Flink's FileSystem.
> {code:sql}
> -- before
> COMPILE PLAN FOR '/tmp/foo/bar.json'
> EXECUTE PLAN FOR '/tmp/foo/bar.json'
> -- after
> COMPILE PLAN FOR 'file:///tmp/foo/bar.json'
> COMPILE PLAN FOR 'hdfs:///tmp/foo/bar.json'
> COMPILE PLAN FOR 's3:///tmp/foo/bar.json'
> COMPILE PLAN FOR 'oss:///tmp/foo/bar.json'
> EXECUTE PLAN FOR 'file:///tmp/foo/bar.json'
> EXECUTE PLAN FOR 'hdfs:///tmp/foo/bar.json'
> EXECUTE PLAN FOR 's3:///tmp/foo/bar.json'
> EXECUTE PLAN FOR 'oss:///tmp/foo/bar.json' {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-31956) Extend the CompiledPlan to read from/write to Flink's FileSystem
Jane Chan created FLINK-31956:
-
Summary: Extend the CompiledPlan to read from/write to Flink's
FileSystem
Key: FLINK-31956
URL: https://issues.apache.org/jira/browse/FLINK-31956
Project: Flink
Issue Type: Sub-task
Components: Table SQL / Client, Table SQL / Planner
Affects Versions: 1.18.0
Reporter: Jane Chan
Fix For: 1.18.0
At present, COMPILE/EXECUTE PLAN FOR '${plan.json}' only supports writing
to/reading from a local file without the scheme. We propose to extend the
support for Flink's FileSystem.
{code:java}
-- before
COMPILE PLAN FOR '/tmp/foo/bar.json'
EXECUTE PLAN FOR '/tmp/foo/bar.json'
-- after
COMPILE PLAN FOR 'file:///tmp/foo/bar.json'
COMPILE PLAN FOR 'hdfs:///tmp/foo/bar.json'
COMPILE PLAN FOR 's3:///tmp/foo/bar.json'
COMPILE PLAN FOR 'oss:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'file:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'hdfs:///tmp/foo/bar.json'
EXECUTE PLAN FOR 's3:///tmp/foo/bar.json'
EXECUTE PLAN FOR 'oss:///tmp/foo/bar.json' {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-31952) Support 'EXPLAIN' statement for CompiledPlan
[
https://issues.apache.org/jira/browse/FLINK-31952?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17717033#comment-17717033
]
Jane Chan commented on FLINK-31952:
---
Hi [~xu_shuai_], thanks for your interest in this ticket! The brief sounds good
to me. cc [~luoyuxia]
> Support 'EXPLAIN' statement for CompiledPlan
>
>
> Key: FLINK-31952
> URL: https://issues.apache.org/jira/browse/FLINK-31952
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: 1.18.0
>
>
> Support the explain SQL syntax towards serialized CompiledPlan
> {code:sql}
> EXPLAIN [ | PLAN FOR]
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31952) Support 'EXPLAIN' statement for CompiledPlan
[
https://issues.apache.org/jira/browse/FLINK-31952?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31952:
--
Description:
Support the explain SQL syntax towards serialized CompiledPlan
{code:sql}
EXPLAIN [ | PLAN FOR]
{code}
> Support 'EXPLAIN' statement for CompiledPlan
>
>
> Key: FLINK-31952
> URL: https://issues.apache.org/jira/browse/FLINK-31952
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: 1.18.0
>
>
> Support the explain SQL syntax towards serialized CompiledPlan
> {code:sql}
> EXPLAIN [ | PLAN FOR]
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-31952) Support 'EXPLAIN' statement for CompiledPlan
Jane Chan created FLINK-31952: - Summary: Support 'EXPLAIN' statement for CompiledPlan Key: FLINK-31952 URL: https://issues.apache.org/jira/browse/FLINK-31952 Project: Flink Issue Type: Sub-task Components: Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31950) Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan
[ https://issues.apache.org/jira/browse/FLINK-31950?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31950: -- Description: According to the FLIP design, we're about to introduce * StateMetadata, which describes the TTL attribute of the stateful stream operator. * StateMetadata SerDerializers. * For StreamExecNodes which translate to stateful and TTL-sensitive operators, change #translateToPlanInternal on how to fetch the TTL configuration during plan deserialization. > Introduce StateMetadata and support operator-level state TTL configuration > via Compiled Plan > > > Key: FLINK-31950 > URL: https://issues.apache.org/jira/browse/FLINK-31950 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Priority: Major > Fix For: 1.18.0 > > > According to the FLIP design, we're about to introduce > * StateMetadata, which describes the TTL attribute of the stateful stream > operator. > * StateMetadata SerDerializers. > * For StreamExecNodes which translate to stateful and TTL-sensitive > operators, change #translateToPlanInternal on how to fetch the TTL > configuration during plan deserialization. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-31950) Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan
Jane Chan created FLINK-31950: - Summary: Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan Key: FLINK-31950 URL: https://issues.apache.org/jira/browse/FLINK-31950 Project: Flink Issue Type: Sub-task Components: Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31791) FLIP-292: Enhance COMPILED PLAN to support operator-level state TTL configuration
[ https://issues.apache.org/jira/browse/FLINK-31791?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31791: -- Description: This is the parent task of FLIP-292. > FLIP-292: Enhance COMPILED PLAN to support operator-level state TTL > configuration > - > > Key: FLINK-31791 > URL: https://issues.apache.org/jira/browse/FLINK-31791 > Project: Flink > Issue Type: New Feature > Components: Table SQL / Planner >Affects Versions: 1.18.0 >Reporter: Jane Chan >Priority: Major > > This is the parent task of FLIP-292. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31917) Loss of Idempotence in JsonSerDe Round Trip for AggregateCall and RexNode
[
https://issues.apache.org/jira/browse/FLINK-31917?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31917:
--
Description:
JsonSerDeTestUtil#testJsonRoundTrip only checks the equality between spec and
deserialized object. Some corner cases are detected when serializing the
deserialized object again.
{code:java}
static T testJsonRoundTrip(SerdeContext serdeContext, T spec, Class
clazz)
throws IOException {
String actualJson = toJson(serdeContext, spec);
T actual = toObject(serdeContext, actualJson, clazz);
assertThat(actual).isEqualTo(spec);
assertThat(actualJson).isEqualTo(toJson(serdeContext, actual)); // this
will eval some corner cases
return actual;
}
{code}
The discovered corner cases are listed as follows.
h5. 1. SerDe for AggregateCall
When deserializing the aggregate call, we should check the JsonNodeType to
avoid converting null to "null" string.
[https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AggregateCallJsonDeserializer.java#L64]
h5. Suggested Fix
{code:java}
JsonNode nameNode = jsonNode.required(FIELD_NAME_NAME);
final String name = JsonNodeType.NULL ? null : nameNode.asText();
{code}
h5. 2. SerDe for RexNode
RexNodeJsonSerdeTest#testSystemFunction should create the temporary system
function instead of the temporary catalog function.
[https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerdeTest.java#L209]
h5. Suggested Fix
Use functionCatalog#registerTemporarySystemFunction to test.
h5. 3. About RexLiteral type
RexNodeJsonSerdeTest#testRexNodeSerde has a test spec as follows
{code:java}
//This will create the literal with DOUBLE as the literal type, and DECIMAL as
the broad type of this literal. You can refer to Calcite for more details
rexBuilder.makeExactLiteral(BigDecimal.valueOf(Double.MAX_VALUE),
FACTORY.createSqlType(SqlTypeName.DOUBLE))
{code}
The RexNodeJsonSerializer uses `typeName`(which is DECIMAL) as the literal's
type, as a result, the rel data type is serialized as double, but the value is
serialized as a string (in case lost the precision)
[https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerializer.java#L197]
And then, during the deserialization, according to the JSON, the deserialized
literal will assign DOUBLE as the literal type and the broad type of the
literal.
This will cause the comparison failure
{code:java}
expected: {"kind": "LITERAL", "value": "1.7976931348623157E+308"}
actual: {"kind": "LITERAL", "value": 1.7976931348623157E+308}
{code}
h5. Suggested Fix
SARG is a special case and can be coped first, and for the rest type, we can
use literal.getType().getSqlTypeName() instead of literal.getTypeName().
{code:java}
// first cope with SARG type
if (literal.getTypeName() == SARG) {
serializeSargValue(
(Sarg) value, literal.getType().getSqlTypeName(), gen,
serializerProvider);
} else {
serializeLiteralValue(
value,
literal.getType().getSqlTypeName(),
gen,
serializerProvider);
}
{code}
was:
JsonSerDeTestUtil#testJsonRoundTrip only checks the equality between spec and
deserialized object. Some corner cases are detected when serializing the
deserialized object again.
{code:java}
static T testJsonRoundTrip(SerdeContext serdeContext, T spec, Class
clazz)
throws IOException {
String actualJson = toJson(serdeContext, spec);
T actual = toObject(serdeContext, actualJson, clazz);
assertThat(actual).isEqualTo(spec);
assertThat(actualJson).isEqualTo(toJson(serdeContext, actual)); // this
will eval some corner cases
return actual;
}
{code}
The discovered corner cases are listed as follows.
h5. 1. SerDe for AggregateCall
When deserializing the aggregate call, we should check the JsonNodeType to
avoid converting null to "null" string.
[https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AggregateCallJsonDeserializer.java#L64]
h5. Suggested Fix
{code:java}
JsonNode nameNode = jsonNode.required(FIELD_NAME_NAME);
final String name = JsonNodeType.NULL ? null : nameNode.asText();
{code}
h5. 2. SerDe for RexNode
RexNodeJsonSerdeTest#testSystemFunction should create the temporary system
function instead of the temporary catalog function.
[jira] [Updated] (FLINK-31917) Loss of Idempotence in JsonSerDe Round Trip for AggregateCall and RexNode
[
https://issues.apache.org/jira/browse/FLINK-31917?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31917:
--
Description:
JsonSerDeTestUtil#testJsonRoundTrip only checks the equality between spec and
deserialized object. Some corner cases are detected when serializing the
deserialized object again.
{code:java}
static T testJsonRoundTrip(SerdeContext serdeContext, T spec, Class
clazz)
throws IOException {
String actualJson = toJson(serdeContext, spec);
T actual = toObject(serdeContext, actualJson, clazz);
assertThat(actual).isEqualTo(spec);
assertThat(actualJson).isEqualTo(toJson(serdeContext, actual)); // this
will eval some corner cases
return actual;
}
{code}
The discovered corner cases are listed as follows.
h5. 1. SerDe for AggregateCall
When deserializing the aggregate call, we should check the JsonNodeType to
avoid converting null to "null" string.
[https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AggregateCallJsonDeserializer.java#L64]
h5. Suggested Fix
{code:java}
JsonNode nameNode = jsonNode.required(FIELD_NAME_NAME);
final String name = JsonNodeType.NULL ? null : nameNode.asText();
{code}
h5. 2. SerDe for RexNode
RexNodeJsonSerdeTest#testSystemFunction should create the temporary system
function instead of the temporary catalog function.
[https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerdeTest.java#L209]
h5. 3. About RexLiteral type
RexNodeJsonSerdeTest#testRexNodeSerde has a test spec as follows
{code:java}
//This will create the literal with DOUBLE as the literal type, and DECIMAL as
the broad type of this literal. You can refer to Calcite for more details
rexBuilder.makeExactLiteral(BigDecimal.valueOf(Double.MAX_VALUE),
FACTORY.createSqlType(SqlTypeName.DOUBLE))
{code}
The RexNodeJsonSerializer uses `typeName`(which is DECIMAL) as the literal's
type, as a result, the rel data type is serialized as double, but the value is
serialized as a string (in case lost the precision)
[https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerializer.java#L197]
And then, during the deserialization, according to the JSON, the deserialized
literal will assign DOUBLE as the literal type and the broad type of the
literal.
This will cause the comparison failure
{code}
expected: {"kind": "LITERAL", "value": "1.7976931348623157E+308"}
actual: {"kind": "LITERAL", "value": 1.7976931348623157E+308}
{code}
> Loss of Idempotence in JsonSerDe Round Trip for AggregateCall and RexNode
> -
>
> Key: FLINK-31917
> URL: https://issues.apache.org/jira/browse/FLINK-31917
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.18.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: 1.18.0
>
>
> JsonSerDeTestUtil#testJsonRoundTrip only checks the equality between spec and
> deserialized object. Some corner cases are detected when serializing the
> deserialized object again.
> {code:java}
> static T testJsonRoundTrip(SerdeContext serdeContext, T spec, Class
> clazz)
> throws IOException {
> String actualJson = toJson(serdeContext, spec);
> T actual = toObject(serdeContext, actualJson, clazz);
> assertThat(actual).isEqualTo(spec);
> assertThat(actualJson).isEqualTo(toJson(serdeContext, actual)); //
> this will eval some corner cases
> return actual;
> }
> {code}
> The discovered corner cases are listed as follows.
> h5. 1. SerDe for AggregateCall
> When deserializing the aggregate call, we should check the JsonNodeType to
> avoid converting null to "null" string.
> [https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AggregateCallJsonDeserializer.java#L64]
> h5. Suggested Fix
> {code:java}
> JsonNode nameNode = jsonNode.required(FIELD_NAME_NAME);
> final String name = JsonNodeType.NULL ? null : nameNode.asText();
> {code}
> h5. 2. SerDe for RexNode
> RexNodeJsonSerdeTest#testSystemFunction should create the temporary system
> function instead of the temporary catalog function.
> [https://github.com/apache/flink/blob/f9b3e0b7bc0432001b4a197539a0712b16e0b33b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerdeTest.java#L209]
[jira] [Created] (FLINK-31917) Loss of Idempotence in JsonSerDe Round Trip for AggregateCall and RexNode
Jane Chan created FLINK-31917: - Summary: Loss of Idempotence in JsonSerDe Round Trip for AggregateCall and RexNode Key: FLINK-31917 URL: https://issues.apache.org/jira/browse/FLINK-31917 Project: Flink Issue Type: Bug Components: Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31884) Upgrade ExecNode to new version causes the old serialized plan failed to pass Json SerDe round trip
[
https://issues.apache.org/jira/browse/FLINK-31884?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31884:
--
Description:
h4. How to Reproduce
Firstly, add a test to dump the compiled plan JSON.
{code:java}
@Test
public void debug() {
tableEnv.executeSql("create table foo (f0 int, f1 string) with ('connector'
= 'datagen')");
tableEnv.executeSql("create table bar (f0 int, f1 string) with ('connector'
= 'print')");
tableEnv.compilePlanSql("insert into bar select * from foo")
.writeToFile(new File("/path/to/debug.json"));
}
{code}
The JSON context is as follows
{code:json}
{
"flinkVersion" : "1.18",
"nodes" : [ {
"id" : 1,
"type" : "stream-exec-table-source-scan_1",
"scanTableSource" : {
"table" : {
"identifier" : "`default_catalog`.`default_database`.`foo`",
"resolvedTable" : {
"schema" : {
"columns" : [ {
"name" : "f0",
"dataType" : "INT"
}, {
"name" : "f1",
"dataType" : "VARCHAR(2147483647)"
} ],
"watermarkSpecs" : [ ]
},
"partitionKeys" : [ ],
"options" : {
"connector" : "datagen"
}
}
}
},
"outputType" : "ROW<`f0` INT, `f1` VARCHAR(2147483647)>",
"description" : "TableSourceScan(table=[[default_catalog, default_database,
foo]], fields=[f0, f1])",
"inputProperties" : [ ]
}, {
"id" : 2,
"type" : "stream-exec-sink_1",
"configuration" : {
"table.exec.sink.keyed-shuffle" : "AUTO",
"table.exec.sink.not-null-enforcer" : "ERROR",
"table.exec.sink.type-length-enforcer" : "IGNORE",
"table.exec.sink.upsert-materialize" : "AUTO"
},
"dynamicTableSink" : {
"table" : {
"identifier" : "`default_catalog`.`default_database`.`bar`",
"resolvedTable" : {
"schema" : {
"columns" : [ {
"name" : "f0",
"dataType" : "INT"
}, {
"name" : "f1",
"dataType" : "VARCHAR(2147483647)"
} ],
"watermarkSpecs" : [ ]
},
"partitionKeys" : [ ],
"options" : {
"connector" : "print"
}
}
}
},
"inputChangelogMode" : [ "INSERT" ],
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
},
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
"outputType" : "ROW<`f0` INT, `f1` VARCHAR(2147483647)>",
"description" : "Sink(table=[default_catalog.default_database.bar],
fields=[f0, f1])"
} ],
"edges" : [ {
"source" : 1,
"target" : 2,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
} ]
}
{code}
Then upgrade the StreamExecSink to a new version
{code:java}
@ExecNodeMetadata(
name = "stream-exec-sink",
version = 1,
consumedOptions = {
"table.exec.sink.not-null-enforcer",
"table.exec.sink.type-length-enforcer",
"table.exec.sink.upsert-materialize",
"table.exec.sink.keyed-shuffle"
},
producedTransformations = {
CommonExecSink.CONSTRAINT_VALIDATOR_TRANSFORMATION,
CommonExecSink.PARTITIONER_TRANSFORMATION,
CommonExecSink.UPSERT_MATERIALIZE_TRANSFORMATION,
CommonExecSink.TIMESTAMP_INSERTER_TRANSFORMATION,
CommonExecSink.SINK_TRANSFORMATION
},
minPlanVersion = FlinkVersion.v1_15,
minStateVersion = FlinkVersion.v1_15)
@ExecNodeMetadata(
name = "stream-exec-sink",
version = 2,
consumedOptions = {
"table.exec.sink.not-null-enforcer",
"table.exec.sink.type-length-enforcer",
"table.exec.sink.upsert-materialize",
"table.exec.sink.keyed-shuffle"
},
producedTransformations = {
CommonExecSink.CONSTRAINT_VALIDATOR_TRANSFORMATION,
CommonExecSink.PARTITIONER_TRANSFORMATION,
CommonExecSink.UPSERT_MATERIALIZE_TRANSFORMATION,
CommonExecSink.TIMESTAMP_INSERTER_TRANSFORMATION,
CommonExecSink.SINK_TRANSFORMATION
},
minPlanVersion = FlinkVersion.v1_18,
minStateVersion = FlinkVersion.v1_15)
public class StreamExecSink extends CommonExecSink implements
StreamExecNode {
}
{code}
And then load the previous plan and print it as JSON text
{code:java}
tableEnv.loadPlan(PlanReference.fromFile("/path/to/debug.json")).printJsonString();
{code}
The SerDe lost idempotence since the version for StreamExecSink became version
2.
{code:json}
{
"flinkVersion" : "1.18",
"nodes" : [ {
"id" : 1,
"type" : "stream-exec-table-source-scan_1",
"scanTableSource" : {
"table" : {
"identifier" : "`default_catalog`.`default_database`.`foo`"
}
[jira] [Created] (FLINK-31884) Upgrade ExecNode to new version causes the old serialized plan failed to pass Json SerDe round trip
Jane Chan created FLINK-31884: - Summary: Upgrade ExecNode to new version causes the old serialized plan failed to pass Json SerDe round trip Key: FLINK-31884 URL: https://issues.apache.org/jira/browse/FLINK-31884 Project: Flink Issue Type: Bug Components: Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan Fix For: 1.18.0 h4. How to Reproduce h4. Root Cause h4. Suggested Fix -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-31830) Coalesce on nested fields with different nullabilities will get wrong plan
[
https://issues.apache.org/jira/browse/FLINK-31830?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17713353#comment-17713353
]
Jane Chan commented on FLINK-31830:
---
I'd like to take this ticket. cc [~lincoln.86xy]
> Coalesce on nested fields with different nullabilities will get wrong plan
> --
>
> Key: FLINK-31830
> URL: https://issues.apache.org/jira/browse/FLINK-31830
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.14.6
>Reporter: lincoln lee
>Priority: Major
>
> A test case similar to FLINK-31829, only changes the nullable field `a.np` to
> not null, will get a wrong plan in 1.14.x (reported from the community user):
> {code}
> @Test
> def testCoalesceOnNestedColumns(): Unit = {
> val tEnv = util.tableEnv
> val tableDescriptor = TableDescriptor.forConnector("datagen")
> .schema(Schema.newBuilder
> .column("id", DataTypes.INT.notNull)
> .column("a", DataTypes.ROW(DataTypes.FIELD("np",
> DataTypes.INT.notNull())).nullable)
> .build)
> .build
> tEnv.createTemporaryTable("t1", tableDescriptor)
> tEnv.createTemporaryTable("t2", tableDescriptor)
> val res = tEnv.executeSql("EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np)
> c1, IFNULL(a.a.np, b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a
> is null or a.a.np is null")
> res.print()
> }
> == Abstract Syntax Tree ==
> LogicalProject(id=[$0], c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])
> +- LogicalFilter(condition=[OR(IS NULL($1), IS NULL(CAST($1.np):INTEGER))])
>+- LogicalJoin(condition=[=($0, $2)], joinType=[left])
> :- LogicalTableScan(table=[[default_catalog, default_database, t1]])
> +- LogicalTableScan(table=[[default_catalog, default_database, t2]])
> {code}
> the top project in the ast is wrong: `LogicalProject(id=[$0],
> c1=[CAST($1.np):INTEGER], c2=[IFNULL($1.np, $3.np)])`, the
> `c1=[CAST($1.np):INTEGER]` relate to `COALESCE(a.a.np, b.a.np) c1` is
> incorrect,
> but this works fine when using sql ddl to create tables
> {code}
> @Test
> def testCoalesceOnNestedColumns2(): Unit = {
> val tEnv = util.tableEnv
> tEnv.executeSql(
> s"""
> |create temporary table t1 (
> | id int not null,
> | a row
> |) with (
> | 'connector' = 'datagen'
> |)
> |""".stripMargin)
> tEnv.executeSql(
> s"""
> |create temporary table t2 (
> | id int not null,
> | a row
> |) with (
> | 'connector' = 'datagen'
> |)
> |""".stripMargin)
> val res = tEnv.executeSql(
> "EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np) c1, IFNULL(a.a.np,
> b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a is null or a.a.np
> is null")
> res.print()
> }
> {code}
> from 1.15, the coalesce will be a new builtin function, and the ast looks
> correct in version 1.15+, while before 1.15 it was rewritten as `case when`
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-31829) Conversion to relational algebra failed to preserve datatypes' nullabilities
[
https://issues.apache.org/jira/browse/FLINK-31829?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17713352#comment-17713352
]
Jane Chan commented on FLINK-31829:
---
Sure. I'd like to take this ticket. I reproduced this issue in the local env
and found that FlinkTypeFactory#createTypeWithNullability only changes the
outer type's nullability for the row type, which is very similar to FLINK-30282.
> Conversion to relational algebra failed to preserve datatypes' nullabilities
> -
>
> Key: FLINK-31829
> URL: https://issues.apache.org/jira/browse/FLINK-31829
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0
>Reporter: lincoln lee
>Priority: Major
> Fix For: 1.18.0
>
>
> AssertionError when run such a case:
> {code}
> @Test
> def testCoalesceOnNestedColumns(): Unit = {
> val tEnv = util.tableEnv
> val tableDescriptor = TableDescriptor
> .forConnector("datagen")
> .schema(
> Schema.newBuilder
> .column("id", DataTypes.INT.notNull)
> .column("a", DataTypes.ROW(DataTypes.FIELD("np",
> DataTypes.INT)).nullable)
> .build)
> .build
> tEnv.createTemporaryTable("t1", tableDescriptor)
> tEnv.createTemporaryTable("t2", tableDescriptor)
> val res = tEnv.executeSql(
> "EXPLAIN SELECT a.id, COALESCE(a.a.np, b.a.np) c1, IFNULL(a.a.np,
> b.a.np) c2 FROM t1 a left JOIN t2 b ON a.id=b.id where a.a is null or a.a.np
> is null")
> res.print()
> }
> {code}
> stack:
> {code}
> java.lang.AssertionError: Conversion to relational algebra failed to preserve
> datatypes:
> validated type:
> RecordType(INTEGER B1, INTEGER NOT NULL B2, INTEGER BenchmarkId1, INTEGER NOT
> NULL BenchmarkIdWithIfNull, INTEGER NOT NULL BenchmarkId2) NOT NULL
> converted type:
> RecordType(INTEGER NOT NULL B1, INTEGER NOT NULL B2, INTEGER BenchmarkId1,
> INTEGER NOT NULL BenchmarkIdWithIfNull, INTEGER NOT NULL BenchmarkId2) NOT
> NULL
> rel:
> LogicalProject(B1=[$4.BenchmarkId], B2=[$2.BenchmarkId], BenchmarkId1=[IF(IS
> NOT NULL($4), $4.BenchmarkId, IF(true, $2.BenchmarkId, null:INTEGER))],
> BenchmarkIdWithIfNull=[IFNULL($4.BenchmarkId, $2.BenchmarkId)],
> BenchmarkId2=[COALESCE($4.BenchmarkId, $2.BenchmarkId)])
> LogicalFilter(condition=[OR(IS NULL($4), IS NULL($4.BenchmarkId))])
> LogicalJoin(condition=[=($3, $0)], joinType=[left])
> LogicalJoin(condition=[=($1, $0)], joinType=[inner])
> LogicalTableScan(table=[[default_catalog, default_database,
> dbo_book]])
> LogicalTableScan(table=[[default_catalog, default_database,
> static_book]])
> LogicalTableScan(table=[[default_catalog, default_database,
> onebook_book_benchmark]])
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.checkConvertedType(SqlToRelConverter.java:500)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:611)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:216)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:192)
> at
> org.apache.flink.table.planner.operations.SqlNodeConvertContext.toRelRoot(SqlNodeConvertContext.java:56)
> at
> org.apache.flink.table.planner.operations.converters.SqlQueryConverter.convertSqlNode(SqlQueryConverter.java:48)
> at
> org.apache.flink.table.planner.operations.converters.SqlNodeConverters.convertSqlNode(SqlNodeConverters.java:65)
> at
> org.apache.flink.table.planner.operations.SqlNodeToOperationConversion.convertValidatedSqlNode(SqlNodeToOperationConversion.java:281)
> at
> org.apache.flink.table.planner.operations.SqlNodeToOperationConversion.convert(SqlNodeToOperationConversion.java:271)
> at
> org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:106)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:665)
> {code}
> but the equivalent tests using sql ddl to create table works fine:
> {code}
> @Test
> def testCoalesceOnNestedColumns2(): Unit = {
> val tEnv = util.tableEnv
> tEnv.executeSql(s"""
>|create temporary table t1 (
>| id int not null,
>| a row
>|) with (
>| 'connector' = 'datagen'
>|)
>|""".stripMargin)
> tEnv.executeSql(s"""
>|create temporary table t2 (
>| id int not null,
>| a row
>|) with (
>|
[jira] [Created] (FLINK-31791) FLIP-292: Enhance COMPILED PLAN to support operator-level state TTL configuration
Jane Chan created FLINK-31791: - Summary: FLIP-292: Enhance COMPILED PLAN to support operator-level state TTL configuration Key: FLINK-31791 URL: https://issues.apache.org/jira/browse/FLINK-31791 Project: Flink Issue Type: New Feature Components: Table SQL / Planner Affects Versions: 1.18.0 Reporter: Jane Chan -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17706313#comment-17706313
]
Jane Chan commented on FLINK-31165:
---
I agree with the error msg improvement, and I'd like to do this task. Cc
[~lincoln.86xy] [~godfrey]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule FlinkLogicalOverAggregateConverter(in:NONE,out:LOGICAL), args
> [rel#245:LogicalWindow.NONE.any.None:
> 0.[NONE].[NONE](input=RelSubset#244,window#0=window(partition {0} rows
> between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))]
> at
> org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
> at
> org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
> at
> org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
> at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:187)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:185)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:189)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:184)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:176)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
> at
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:87)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:315)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:195)
> at
> org.apache.flink.table.api.bridge.internal.AbstractStreamTableEnvironmentImpl.toStreamInternal(AbstractStreamTableEnvironmentImpl.java:224)
> at
> org.apache.flink.table.api.bridge.internal.AbstractStreamTableEnvironmentImpl.toStreamInternal(AbstractStreamTableEnvironmentImpl.java:219)
> at
>
[jira] [Commented] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703655#comment-17703655
]
Jane Chan commented on FLINK-31165:
---
The current behavior under the Flink batch is the same as under streaming mode
since this rewrite rule is applied during the LogicalWindow creation.
My concern mainly comes from the implementation aspect. The reason that caused
this problem is the use of constant folding optimization when creating
LogicalWindow, which leads to the orderByKey being empty when passed to
FlinkLogicalOverAggregateConverter.
There are two possible solutions. The first one is to remove the constant
folding optimization or add some judgment here, such as giving up optimization
when orderByKey becomes empty after optimization. The second one is to remove
the check of orderByKey in FlinkLogicalOverAggregateConverter, but then the
problem becomes how to distinguish between order by constants and no order by
clause.
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule FlinkLogicalOverAggregateConverter(in:NONE,out:LOGICAL), args
> [rel#245:LogicalWindow.NONE.any.None:
> 0.[NONE].[NONE](input=RelSubset#244,window#0=window(partition {0} rows
> between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))]
> at
> org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
> at
> org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
> at
> org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
> at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:187)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:185)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:189)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:184)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:176)
> at
>
[jira] [Commented] (FLINK-31375) the values of map are truncated by the CASE WHEN function.
[
https://issues.apache.org/jira/browse/FLINK-31375?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703513#comment-17703513
]
Jane Chan commented on FLINK-31375:
---
Hi [~jeff-zou], thanks for reporting this issue. I've reproduced it using
Flink-1.15.1, 1.15.2, and 1.16.1
!image-2023-03-22-14-20-18-575.png|width=616,height=214!
This is a bug caused by
[CALCITE-4603|https://issues.apache.org/jira/browse/CALCITE-4603], and has been
fixed in Calcite 1.27.0
As a workaround, you can try manually casting 'abc' as varchar like the
following.
{code:sql}
select * from (values(case when true then map['test','123456789'] else map
['msg_code','0', 'msg_reason', cast('abc' as string)] end)); {code}
As release-1.17 has upgraded to Calcite 1.29.0, this issue has been fixed.
> the values of map are truncated by the CASE WHEN function.
> -
>
> Key: FLINK-31375
> URL: https://issues.apache.org/jira/browse/FLINK-31375
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API, Table SQL / Runtime
>Affects Versions: 1.15.1
>Reporter: jeff-zou
>Priority: Major
> Attachments: image-2023-03-22-14-20-18-575.png
>
>
> the values of map are truncated by the CASE WHEN function.
> {code:java}
> // sql
> create table test (a map) with ('connector'='print');
> insert into test select * from (values(case when true then
> map['test','123456789'] else map ['msg_code','0', 'msg_reason', 'abc']
> end));{code}
> the result:
> {code:java}
> +I[{test=123}] {code}
> We hope the value of result is '123456789', but I get '123', the length is
> limited by 'abc'.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31375) the values of map are truncated by the CASE WHEN function.
[
https://issues.apache.org/jira/browse/FLINK-31375?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31375:
--
Attachment: image-2023-03-22-14-20-18-575.png
> the values of map are truncated by the CASE WHEN function.
> -
>
> Key: FLINK-31375
> URL: https://issues.apache.org/jira/browse/FLINK-31375
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API, Table SQL / Runtime
>Affects Versions: 1.15.1
>Reporter: jeff-zou
>Priority: Major
> Attachments: image-2023-03-22-14-20-18-575.png
>
>
> the values of map are truncated by the CASE WHEN function.
> {code:java}
> // sql
> create table test (a map) with ('connector'='print');
> insert into test select * from (values(case when true then
> map['test','123456789'] else map ['msg_code','0', 'msg_reason', 'abc']
> end));{code}
> the result:
> {code:java}
> +I[{test=123}] {code}
> We hope the value of result is '123456789', but I get '123', the length is
> limited by 'abc'.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703466#comment-17703466
]
Jane Chan edited comment on FLINK-31165 at 3/22/23 6:04 AM:
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite. See
[LogicalWindow.java#L371|https://github.com/apache/flink/blob/268fc1a46f8af171c7102229a010af71c56623d0/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java#L371],
when applying Calcite's ProjectToWindowRule.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by key
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, SQLServer, Spark, and Hive.
SQLSrver will throw an exception that "Windowed functions and NEXT VALUE FOR
functions do not support constants as ORDER BY clause expressions."; the rest
do allow this to happen, and just output the first inserted row. Do you think
we need to align this behavior? Or at least throwing a more meaningful error.
cc [~godfreyhe] [~lincoln.86xy]
was (Author: qingyue):
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite. See
[LogicalWindow.java#L371|https://github.com/apache/flink/blob/268fc1a46f8af171c7102229a010af71c56623d0/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java#L371],
when applying Calcite's ProjectToWindowRule.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, Spark, and Hive, they do
allow this to happen, and just output the first inserted row. Do you think we
need to align this behavior? Or at least throwing a more meaningful error. cc
[~godfreyhe] [~lincoln.86xy]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule
[jira] [Comment Edited] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703466#comment-17703466
]
Jane Chan edited comment on FLINK-31165 at 3/22/23 6:04 AM:
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite. See
[LogicalWindow.java#L371|https://github.com/apache/flink/blob/268fc1a46f8af171c7102229a010af71c56623d0/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java#L371],
when applying Calcite's ProjectToWindowRule.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by key
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, SQLServer, Spark, and Hive.
SQLServer will throw an exception that "Windowed functions and NEXT VALUE FOR
functions do not support constants as ORDER BY clause expressions."; the rest
do allow this to happen, and just output the first inserted row. Do you think
we need to align this behavior? Or at least throwing a more meaningful error.
cc [~godfreyhe] [~lincoln.86xy]
was (Author: qingyue):
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite. See
[LogicalWindow.java#L371|https://github.com/apache/flink/blob/268fc1a46f8af171c7102229a010af71c56623d0/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java#L371],
when applying Calcite's ProjectToWindowRule.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by key
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, SQLServer, Spark, and Hive.
SQLSrver will throw an exception that "Windowed functions and NEXT VALUE FOR
functions do not support constants as ORDER BY clause expressions."; the rest
do allow this to happen, and just output the first inserted row. Do you think
we need to align this behavior? Or at least throwing a more meaningful error.
cc [~godfreyhe] [~lincoln.86xy]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job
[jira] [Commented] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703499#comment-17703499
]
Jane Chan commented on FLINK-31165:
---
I rethink it, from the streaming semantics the result might be
non-deterministic if supporting order by constants. So I suggest throwing a
meaningful error to indicate users not to use constants as the order by key.
WDYT? [~godfrey] [~lincoln.86xy]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule FlinkLogicalOverAggregateConverter(in:NONE,out:LOGICAL), args
> [rel#245:LogicalWindow.NONE.any.None:
> 0.[NONE].[NONE](input=RelSubset#244,window#0=window(partition {0} rows
> between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))]
> at
> org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
> at
> org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
> at
> org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
> at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:187)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:185)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:189)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:184)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:176)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
> at
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:87)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:315)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:195)
> at
> org.apache.flink.table.api.bridge.internal.AbstractStreamTableEnvironmentImpl.toStreamInternal(AbstractStreamTableEnvironmentImpl.java:224)
> at
>
[jira] [Comment Edited] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703466#comment-17703466
]
Jane Chan edited comment on FLINK-31165 at 3/22/23 4:29 AM:
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite. See
[LogicalWindow.java#L371|https://github.com/apache/flink/blob/268fc1a46f8af171c7102229a010af71c56623d0/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java#L371],
when applying Calcite's ProjectToWindowRule.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, Spark, and Hive, they do
allow this to happen, and just output the first inserted row. Do you think we
need to align this behavior? Or at least throwing a more meaningful error. cc
[~godfreyhe] [~lincoln.86xy]
was (Author: qingyue):
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite. See
[LogicalWindow.java#L371|https://github.com/apache/flink/blob/268fc1a46f8af171c7102229a010af71c56623d0/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java#L371],
when applying Calcite's ProjectToWindowRule.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, Spark, and Hive, they do
allow this to happen, and just output the first inserted row. Do you think we
need to align this behavior? cc [~godfreyhe] [~lincoln.86xy]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule FlinkLogicalOverAggregateConverter(in:NONE,out:LOGICAL), args
> [rel#245:LogicalWindow.NONE.any.None:
> 0.[NONE].[NONE](input=RelSubset#244,window#0=window(partition {0} rows
> between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))]
>
[jira] [Comment Edited] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703466#comment-17703466
]
Jane Chan edited comment on FLINK-31165 at 3/22/23 4:28 AM:
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite. See
[LogicalWindow.java#L371|https://github.com/apache/flink/blob/268fc1a46f8af171c7102229a010af71c56623d0/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java#L371],
when applying Calcite's ProjectToWindowRule.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, Spark, and Hive, they do
allow this to happen, and just output the first inserted row. Do you think we
need to align this behavior? cc [~godfreyhe] [~lincoln.86xy]
was (Author: qingyue):
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, Spark, and Hive, they do
allow this to happen, and just output the first inserted row. Do you think we
need to align this behavior? cc [~godfreyhe] [~lincoln.86xy]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule FlinkLogicalOverAggregateConverter(in:NONE,out:LOGICAL), args
> [rel#245:LogicalWindow.NONE.any.None:
> 0.[NONE].[NONE](input=RelSubset#244,window#0=window(partition {0} rows
> between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))]
> at
> org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
> at
> org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
> at
> org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
> at
[jira] [Comment Edited] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703466#comment-17703466
]
Jane Chan edited comment on FLINK-31165 at 3/22/23 3:49 AM:
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL, PostgreSQL, Spark, and Hive, they do
allow this to happen, and just output the first inserted row. Do you think we
need to align this behavior? cc [~godfreyhe] [~lincoln.86xy]
was (Author: qingyue):
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL and PostgreSQL, they do allow this to
happen, and just output the first inserted row. Do you think we need to align
this behavior? cc [~godfreyhe] [~lincoln.86xy]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule FlinkLogicalOverAggregateConverter(in:NONE,out:LOGICAL), args
> [rel#245:LogicalWindow.NONE.any.None:
> 0.[NONE].[NONE](input=RelSubset#244,window#0=window(partition {0} rows
> between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))]
> at
> org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
> at
> org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
> at
> org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
> at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62)
> at
>
[jira] [Commented] (FLINK-31165) Over Agg: The window rank function without order by error in top N query
[
https://issues.apache.org/jira/browse/FLINK-31165?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17703466#comment-17703466
]
Jane Chan commented on FLINK-31165:
---
Hi [~rohankrao] , thanks for reporting this issue.
Actually, the order by field SRC_NO is a constant and is folded during query
rewrite.
{code:java}
LogicalProject(inputs=[0..2])
+- LogicalFilter(condition=[<=($2, 1)])
+- LogicalProject(inputs=[0..1], exprs=[[ROW_NUMBER() OVER (PARTITION BY $0
ORDER BY 2022-01-01 NULLS FIRST)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
temp_table]]) {code}
>From the perspective of SQL semantics, using a constant as the order by field
>for row_number has no meaning, because the constant will not change the
>sorting result of row_number, and each row will get the same rank. As a
>current workaround, please try to specify another field to ensure that
>row_number is sorted in the correct order.
While I tested the query against MySQL and PostgreSQL, they do allow this to
happen, and just output the first inserted row. Do you think we need to align
this behavior? cc [~godfreyhe] [~lincoln.86xy]
> Over Agg: The window rank function without order by error in top N query
>
>
> Key: FLINK-31165
> URL: https://issues.apache.org/jira/browse/FLINK-31165
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.16.0
>Reporter: P Rohan Kumar
>Priority: Major
>
>
> {code:java}
> val env: StreamExecutionEnvironment =
> StreamExecutionEnvironment.getExecutionEnvironment
> val tableEnv = StreamTableEnvironment.create(env)
> val td = TableDescriptor.forConnector("datagen").option("rows-per-second",
> "10")
> .option("number-of-rows", "10")
> .schema(Schema
> .newBuilder()
> .column("NAME", DataTypes.VARCHAR(2147483647))
> .column("ROLLNO", DataTypes.DECIMAL(5, 0))
> .column("DOB", DataTypes.DATE())
> .column("CLASS", DataTypes.DECIMAL(2, 0))
> .column("SUBJECT", DataTypes.VARCHAR(2147483647))
> .build())
> .build()
> val table = tableEnv.from(td)
> tableEnv.createTemporaryView("temp_table", table)
> val newTable = tableEnv.sqlQuery("select temp_table.*,cast('2022-01-01' as
> date) SRC_NO from temp_table")
> tableEnv.createTemporaryView("temp_table2", newTable)
> val newTable2 = tableEnv.sqlQuery("select * from (select
> NAME,ROLLNO,row_number() over (partition by NAME ORDER BY SRC_NO) AS rownum
> from temp_table2 a) where rownum <= 1")
> tableEnv.toChangelogStream(newTable2).print()
> env.execute()
> {code}
>
>
> I am getting the below error if I run the above code.
> I have already provided an order by column.
> If I change the order by column to some other column, such as "SUBJECT", then
> the job runs fine.
>
>
> {code:java}
> Exception in thread "main" java.lang.RuntimeException: Error while applying
> rule FlinkLogicalOverAggregateConverter(in:NONE,out:LOGICAL), args
> [rel#245:LogicalWindow.NONE.any.None:
> 0.[NONE].[NONE](input=RelSubset#244,window#0=window(partition {0} rows
> between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))]
> at
> org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
> at
> org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
> at
> org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
> at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:187)
> at
> scala.collection.TraversableOnce$folder$1$.apply(TraversableOnce.scala:185)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:189)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:184)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55)
> at
>
[jira] [Commented] (FLINK-31424) NullPointer when using StatementSet for multiple sinks
[
https://issues.apache.org/jira/browse/FLINK-31424?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17702977#comment-17702977
]
Jane Chan commented on FLINK-31424:
---
I opened a PR, and it would be great if you could help to review it. cc
[~lincoln.86xy] [~zhengyiweng]
> NullPointer when using StatementSet for multiple sinks
> --
>
> Key: FLINK-31424
> URL: https://issues.apache.org/jira/browse/FLINK-31424
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.16.1
>Reporter: jirawech.s
>Assignee: Jane Chan
>Priority: Major
> Labels: pull-request-available
> Attachments: HelloFlinkWindowJoin.java
>
>
> I got following error when i tried to execute multiple sinks using
> StatementSet. I am not sure what it is and where to find possible solution.
> Error
> {code:java}
> // code placeholder
> Exception in thread "main" java.lang.NullPointerException
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMdWindowProperties.getJoinWindowProperties(FlinkRelMdWindowProperties.scala:373)
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMdWindowProperties.getWindowProperties(FlinkRelMdWindowProperties.scala:349)
> at
> GeneratedMetadataHandler_WindowProperties.getWindowProperties_$(Unknown
> Source)
> at GeneratedMetadataHandler_WindowProperties.getWindowProperties(Unknown
> Source)
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery.getRelWindowProperties(FlinkRelMetadataQuery.java:261)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.createIntermediateRelTable(StreamCommonSubGraphBasedOptimizer.scala:287)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeBlock(StreamCommonSubGraphBasedOptimizer.scala:138)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.$anonfun$optimizeBlock$1(StreamCommonSubGraphBasedOptimizer.scala:111)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.$anonfun$optimizeBlock$1$adapted(StreamCommonSubGraphBasedOptimizer.scala:109)
> at
> scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:58)
> at
> scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:51)
> at
> scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47){code}
> You can test the code, please see the attachment
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-31424) NullPointer when using StatementSet for multiple sinks
[
https://issues.apache.org/jira/browse/FLINK-31424?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17700657#comment-17700657
]
Jane Chan edited comment on FLINK-31424 at 3/15/23 12:38 PM:
-
Hi, [~jirawech.s], thanks for reporting this issue.
When multiple sinks are involved, the optimizer will try to break the DAG into
rel blocks. According to the provided SQL, the optimized intermediate
relNode(root) is the `StreamPhysicalWindowJoin`. It is wrapped as the new
table scan source by
`StreamCommonSubGraphBasedOptimizer#createIntermediateRelTable`.
The NPE is thrown because the `FlinkRelMdWindowProperties` does not support
`StreamPhysicalGlobalWindowAggregate`. (Here, the input of
`StreamPhysicalExchange` is `StreamPhysicalGlobalWindowAggregate`).
I would like to fix it. cc [~lincoln.86xy]
was (Author: qingyue):
Hi, [~jirawech.s], thanks for reporting this issue.
When multiple sinks are involved, the optimizer will try to break the DAG into
rel blocks. According to the provided SQL, the optimized intermediate
relNode(root) is the `StreamPhysicalWindowJoin`. It is wrapped as the new
table scan source by
`StreamCommonSubGraphBasedOptimizer#createIntermediateRelTable`.
The NPE is thrown because the `FlinkRelMdWindowProperties` does not support
`StreamPhysicalGlobalWindowAggregate`. (Here, the input of
`StreamPhysicalExChange` is `StreamPhysicalGlobalWindowAggregate`).
I would like to fix it. cc [~lincoln.86xy]
> NullPointer when using StatementSet for multiple sinks
> --
>
> Key: FLINK-31424
> URL: https://issues.apache.org/jira/browse/FLINK-31424
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.16.1
>Reporter: jirawech.s
>Priority: Major
> Attachments: HelloFlinkWindowJoin.java
>
>
> I got following error when i tried to execute multiple sinks using
> StatementSet. I am not sure what it is and where to find possible solution.
> Error
> {code:java}
> // code placeholder
> Exception in thread "main" java.lang.NullPointerException
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMdWindowProperties.getJoinWindowProperties(FlinkRelMdWindowProperties.scala:373)
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMdWindowProperties.getWindowProperties(FlinkRelMdWindowProperties.scala:349)
> at
> GeneratedMetadataHandler_WindowProperties.getWindowProperties_$(Unknown
> Source)
> at GeneratedMetadataHandler_WindowProperties.getWindowProperties(Unknown
> Source)
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery.getRelWindowProperties(FlinkRelMetadataQuery.java:261)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.createIntermediateRelTable(StreamCommonSubGraphBasedOptimizer.scala:287)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeBlock(StreamCommonSubGraphBasedOptimizer.scala:138)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.$anonfun$optimizeBlock$1(StreamCommonSubGraphBasedOptimizer.scala:111)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.$anonfun$optimizeBlock$1$adapted(StreamCommonSubGraphBasedOptimizer.scala:109)
> at
> scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:58)
> at
> scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:51)
> at
> scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47){code}
> You can test the code, please see the attachment
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-31424) NullPointer when using StatementSet for multiple sinks
[
https://issues.apache.org/jira/browse/FLINK-31424?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17700657#comment-17700657
]
Jane Chan commented on FLINK-31424:
---
Hi, [~jirawech.s], thanks for reporting this issue.
When multiple sinks are involved, the optimizer will try to break the DAG into
rel blocks. According to the provided SQL, the optimized intermediate
relNode(root) is the `StreamPhysicalWindowJoin`. It is wrapped as the new
table scan source by
`StreamCommonSubGraphBasedOptimizer#createIntermediateRelTable`.
The NPE is thrown because the `FlinkRelMdWindowProperties` does not support
`StreamPhysicalGlobalWindowAggregate`. (Here, the input of
`StreamPhysicalExChange` is `StreamPhysicalGlobalWindowAggregate`).
I would like to fix it. cc [~lincoln.86xy]
> NullPointer when using StatementSet for multiple sinks
> --
>
> Key: FLINK-31424
> URL: https://issues.apache.org/jira/browse/FLINK-31424
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.16.1
>Reporter: jirawech.s
>Priority: Major
> Attachments: HelloFlinkWindowJoin.java
>
>
> I got following error when i tried to execute multiple sinks using
> StatementSet. I am not sure what it is and where to find possible solution.
> Error
> {code:java}
> // code placeholder
> Exception in thread "main" java.lang.NullPointerException
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMdWindowProperties.getJoinWindowProperties(FlinkRelMdWindowProperties.scala:373)
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMdWindowProperties.getWindowProperties(FlinkRelMdWindowProperties.scala:349)
> at
> GeneratedMetadataHandler_WindowProperties.getWindowProperties_$(Unknown
> Source)
> at GeneratedMetadataHandler_WindowProperties.getWindowProperties(Unknown
> Source)
> at
> org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery.getRelWindowProperties(FlinkRelMetadataQuery.java:261)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.createIntermediateRelTable(StreamCommonSubGraphBasedOptimizer.scala:287)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeBlock(StreamCommonSubGraphBasedOptimizer.scala:138)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.$anonfun$optimizeBlock$1(StreamCommonSubGraphBasedOptimizer.scala:111)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.$anonfun$optimizeBlock$1$adapted(StreamCommonSubGraphBasedOptimizer.scala:109)
> at
> scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:58)
> at
> scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:51)
> at
> scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47){code}
> You can test the code, please see the attachment
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-26051) one sql has row_number =1 and the subsequent SQL has "case when" and "where" statement result Exception : The window can only be ordered in ASCENDING mode
[
https://issues.apache.org/jira/browse/FLINK-26051?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17699929#comment-17699929
]
Jane Chan edited comment on FLINK-26051 at 3/14/23 7:11 AM:
[~KristoffSC], Sorry for the late reply. Feel free to go ahead.
Here are some investigations I made before; I hope that will help.
By changing the CommonCalc#computeSelfCost to
{code:scala}
planner.getCostFactory.makeCost(newRowCnt, newRowCnt * (compCnt + 1), 0) {code}
This issue can be fixed.
Nonetheless, it will cause ~100 test plans to be changed. Over 50% of plans
removed the Calc node that only contains projection. A few tests failed to
generate the plan after applying this change.
I think the leading blocker is evaluating the impact of this change to ensure
there will be no performance regression for the affected tests.
was (Author: qingyue):
[~KristoffSC], Sorry for the late reply. Feel free to go ahead.
Here are some investigations I made before; I hope that will help.
By changing the CommonCalc#computeSelfCost to
{code:scala}
planner.getCostFactory.makeCost(newRowCnt, newRowCnt * (compCnt + 1), 0) {code}
This issue can be fixed.
Nonetheless, it will cause ~100 test plans to be changed. Over 50% of plans
removed the Calc node that only contains projection. A few plans failed after
applying this change.
I think the leading blocker is evaluating the impact of this change to ensure
there will be no performance regression for the affected tests.
> one sql has row_number =1 and the subsequent SQL has "case when" and "where"
> statement result Exception : The window can only be ordered in ASCENDING mode
> --
>
> Key: FLINK-26051
> URL: https://issues.apache.org/jira/browse/FLINK-26051
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.2, 1.14.4
>Reporter: chuncheng wu
>Assignee: Jane Chan
>Priority: Major
> Attachments: image-2022-02-10-20-13-14-424.png,
> image-2022-02-11-11-18-20-594.png, image-2022-06-17-21-28-54-886.png
>
>
> hello,
> i have 2 sqls. One sql (sql0) is "select xx from ( ROW_NUMBER statment)
> where rn=1" and the other one (sql1) is "s{color:#505f79}elect ${fields}
> from result where ${filter_conditions}{color}" . The fields quoted in sql1
> has one "case when" field .The two sql can work well seperately.but if they
> combine it results the exception as follow . It happen in the occasion when
> logical plan turn into physical plan :
>
> {code:java}
> org.apache.flink.table.api.TableException: The window can only be ordered in
> ASCENDING mode.
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.scala:98)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.scala:52)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:59)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregateBase.translateToPlan(StreamExecOverAggregateBase.scala:42)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:54)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:39)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:59)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:66)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:65)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
> at scala.collection.AbstractTraversable.map(Traversable.scala:104)
> at
>
[jira] [Commented] (FLINK-26051) one sql has row_number =1 and the subsequent SQL has "case when" and "where" statement result Exception : The window can only be ordered in ASCENDING mode
[
https://issues.apache.org/jira/browse/FLINK-26051?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17699929#comment-17699929
]
Jane Chan commented on FLINK-26051:
---
[~KristoffSC], Sorry for the late reply. Feel free to go ahead.
Here are some investigations I made before; I hope that will help.
By changing the CommonCalc#computeSelfCost to
{code:scala}
planner.getCostFactory.makeCost(newRowCnt, newRowCnt * (compCnt + 1), 0) {code}
This issue can be fixed.
Nonetheless, it will cause ~100 test plans to be changed. Over 50% of plans
removed the Calc node that only contains projection. A few plans failed after
applying this change.
I think the leading blocker is evaluating the impact of this change to ensure
there will be no performance regression for the affected tests.
> one sql has row_number =1 and the subsequent SQL has "case when" and "where"
> statement result Exception : The window can only be ordered in ASCENDING mode
> --
>
> Key: FLINK-26051
> URL: https://issues.apache.org/jira/browse/FLINK-26051
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.2, 1.14.4
>Reporter: chuncheng wu
>Assignee: Jane Chan
>Priority: Major
> Attachments: image-2022-02-10-20-13-14-424.png,
> image-2022-02-11-11-18-20-594.png, image-2022-06-17-21-28-54-886.png
>
>
> hello,
> i have 2 sqls. One sql (sql0) is "select xx from ( ROW_NUMBER statment)
> where rn=1" and the other one (sql1) is "s{color:#505f79}elect ${fields}
> from result where ${filter_conditions}{color}" . The fields quoted in sql1
> has one "case when" field .The two sql can work well seperately.but if they
> combine it results the exception as follow . It happen in the occasion when
> logical plan turn into physical plan :
>
> {code:java}
> org.apache.flink.table.api.TableException: The window can only be ordered in
> ASCENDING mode.
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.scala:98)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.scala:52)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:59)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregateBase.translateToPlan(StreamExecOverAggregateBase.scala:42)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:54)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:39)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:59)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:66)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:65)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
> at scala.collection.AbstractTraversable.map(Traversable.scala:104)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:65)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner.explain(StreamPlanner.scala:103)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner.explain(StreamPlanner.scala:42)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.explainInternal(TableEnvironmentImpl.java:630)
> at
> org.apache.flink.table.api.internal.TableImpl.explain(TableImpl.java:582)
> at
> com.meituan.grocery.data.flink.test.BugTest.testRowNumber(BugTest.java:69)
> at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native
> Method)
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
> at
>
[jira] [Updated] (FLINK-31404) Support extracting all mapping of Calc node's input references
[ https://issues.apache.org/jira/browse/FLINK-31404?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31404: -- Summary: Support extracting all mapping of Calc node's input references (was: Support extracting source mapping for all RexNode) > Support extracting all mapping of Calc node's input references > -- > > Key: FLINK-31404 > URL: https://issues.apache.org/jira/browse/FLINK-31404 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Affects Versions: 1.17.0, 1.17.1 >Reporter: Jane Chan >Priority: Major > > Currently, `StreamNonDeterministicUpdatePlanVisitor` only supports extracting > a limited kind of RexNode to its input node for the Calc. > * RexInputRef > * RexCall with the operator kind as `AS` and `CAST` > It can be optimized to support all kinds of RexNode by > FlinkRexUtil#findAllInputRefs. > cc [~lincoln.86xy] -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-31404) Support extracting source mapping for all RexNode
Jane Chan created FLINK-31404: - Summary: Support extracting source mapping for all RexNode Key: FLINK-31404 URL: https://issues.apache.org/jira/browse/FLINK-31404 Project: Flink Issue Type: Sub-task Components: Table SQL / Planner Affects Versions: 1.17.0, 1.17.1 Reporter: Jane Chan Currently, `StreamNonDeterministicUpdatePlanVisitor` only supports extracting a limited kind of RexNode to its input node for the Calc. * RexInputRef * RexCall with the operator kind as `AS` and `CAST` It can be optimized to support all kinds of RexNode by FlinkRexUtil#findAllInputRefs. cc [~lincoln.86xy] -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Comment Edited] (FLINK-30640) Unstable test in CliClientITCase
[ https://issues.apache.org/jira/browse/FLINK-30640?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17698849#comment-17698849 ] Jane Chan edited comment on FLINK-30640 at 3/10/23 9:53 AM: I compared the `table.q` file between "expected" v.s. "but was", and found that the only difference is that the table column lengths are different from L#1358~L#1360 and L#1424. cc [~fsk119] Do you have any clue about this? !image-2023-03-10-17-51-42-192.png|width=2634,height=613! !image-2023-03-10-17-49-26-971.png|width=1733,height=214! was (Author: qingyue): I compared the `table.q` file between "expected" v.s. "but was", and found that the only difference is that the table column lengths are different from L#1358~L#1360 and L#1424. !image-2023-03-10-17-51-42-192.png! !image-2023-03-10-17-49-26-971.png! > Unstable test in CliClientITCase > > > Key: FLINK-30640 > URL: https://issues.apache.org/jira/browse/FLINK-30640 > Project: Flink > Issue Type: Bug > Components: Connectors / Hive, Table SQL / Client >Affects Versions: 1.17.0 >Reporter: yuzelin >Assignee: dalongliu >Priority: Blocker > Labels: pull-request-available, test-stability > Fix For: 1.17.0 > > Attachments: image-2023-03-10-17-49-26-971.png, > image-2023-03-10-17-51-42-192.png > > > [https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=44743=logs=0c940707-2659-5648-cbe6-a1ad63045f0a=075c2716-8010-5565-fe08-3c4bb45824a4] > > The failed test can work normally in my local environment. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-30640) Unstable test in CliClientITCase
[ https://issues.apache.org/jira/browse/FLINK-30640?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-30640: -- Attachment: image-2023-03-10-17-51-42-192.png > Unstable test in CliClientITCase > > > Key: FLINK-30640 > URL: https://issues.apache.org/jira/browse/FLINK-30640 > Project: Flink > Issue Type: Bug > Components: Connectors / Hive, Table SQL / Client >Affects Versions: 1.17.0 >Reporter: yuzelin >Assignee: dalongliu >Priority: Blocker > Labels: pull-request-available, test-stability > Fix For: 1.17.0 > > Attachments: image-2023-03-10-17-49-26-971.png, > image-2023-03-10-17-51-42-192.png > > > [https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=44743=logs=0c940707-2659-5648-cbe6-a1ad63045f0a=075c2716-8010-5565-fe08-3c4bb45824a4] > > The failed test can work normally in my local environment. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-30640) Unstable test in CliClientITCase
[ https://issues.apache.org/jira/browse/FLINK-30640?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17698849#comment-17698849 ] Jane Chan commented on FLINK-30640: --- I compared the `table.q` file between "expected" v.s. "but was", and found that the only difference is that the table column lengths are different from L#1358~L#1360 and L#1424. !image-2023-03-10-17-51-42-192.png! !image-2023-03-10-17-49-26-971.png! > Unstable test in CliClientITCase > > > Key: FLINK-30640 > URL: https://issues.apache.org/jira/browse/FLINK-30640 > Project: Flink > Issue Type: Bug > Components: Connectors / Hive, Table SQL / Client >Affects Versions: 1.17.0 >Reporter: yuzelin >Assignee: dalongliu >Priority: Blocker > Labels: pull-request-available, test-stability > Fix For: 1.17.0 > > Attachments: image-2023-03-10-17-49-26-971.png, > image-2023-03-10-17-51-42-192.png > > > [https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=44743=logs=0c940707-2659-5648-cbe6-a1ad63045f0a=075c2716-8010-5565-fe08-3c4bb45824a4] > > The failed test can work normally in my local environment. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-30640) Unstable test in CliClientITCase
[ https://issues.apache.org/jira/browse/FLINK-30640?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-30640: -- Attachment: image-2023-03-10-17-49-26-971.png > Unstable test in CliClientITCase > > > Key: FLINK-30640 > URL: https://issues.apache.org/jira/browse/FLINK-30640 > Project: Flink > Issue Type: Bug > Components: Connectors / Hive, Table SQL / Client >Affects Versions: 1.17.0 >Reporter: yuzelin >Assignee: dalongliu >Priority: Blocker > Labels: pull-request-available, test-stability > Fix For: 1.17.0 > > Attachments: image-2023-03-10-17-49-26-971.png > > > [https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=44743=logs=0c940707-2659-5648-cbe6-a1ad63045f0a=075c2716-8010-5565-fe08-3c4bb45824a4] > > The failed test can work normally in my local environment. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17696532#comment-17696532
]
Jane Chan commented on FLINK-31182:
---
Hi [~twalthr], thanks for the fix! LGTM, and feel free to merge.
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Assignee: Timo Walther
>Priority: Major
> Labels: pull-request-available
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/y6fgzyx330omhkr40376knw8k4oczz3s].
> The stacktrace is
> {code:java}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> the output type strategy is "Missing"; as a result,
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
> !screenshot-1.png|width=675,height=295!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-25986) Add FLIP-190 new API methods to python
[ https://issues.apache.org/jira/browse/FLINK-25986?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17694915#comment-17694915 ] Jane Chan commented on FLINK-25986: --- Is there any update? cc [~nicholasjiang]. If you're not available, I'd like to continue the work. > Add FLIP-190 new API methods to python > -- > > Key: FLINK-25986 > URL: https://issues.apache.org/jira/browse/FLINK-25986 > Project: Flink > Issue Type: Sub-task > Components: API / Python, Table SQL / API >Reporter: Francesco Guardiani >Assignee: Nicholas Jiang >Priority: Major > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Comment Edited] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17692450#comment-17692450
]
Jane Chan edited comment on FLINK-31182 at 2/23/23 6:08 AM:
[~Sergey Nuyanzin] Yes, the check on the output type inference strategy was
introduced in FunctionCatalogTableOperator in FLINK-15487 for FLIP-65 a long
time ago.
I guess it worked well before FLIP-190 was introduced
because`LogicalUnnestRule` will infer the output type and convert UNNEST to
EXPLODE. (see [LogicalUnnestRule
L#99|https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRule.scala#L99]).
As a result, although the BuiltinFunctionDefinition for UNNEST always has a
"MISSING" type inference, it does not affect the execution.
However, after CompiledPlan is introduced, RexNodeJsonDeserializer relies on
the OperatorTable to lookup functions, which always get a static placeholder
(i.e. MISSING) for UNNEST, and due to this check, the deserialization failed.
Could you help to take a look, cc [~godfreyhe] and [~twalthr]
was (Author: qingyue):
[~Sergey Nuyanzin] Yes, the check on the output type inference strategy was
introduced in FunctionCatalogTableOperator in FLINK-15487 for FLIP-65 a long
time ago.
I guess it worked well before FLIP-190 was introduced
because`LogicalUnnestRule` will infer the output type and convert UNNEST to
EXPLODE. (see [LogicalUnnestRule
L#99|https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRule.scala#L99]).
As a result, although the BuiltinFunctionDefinition for UNNEST always has a
"MISSING" type inference, it does not affect the execution.
However, after CompiledPlan is introduced, RexNodeJsonDeserializer relies on
the OperatorTable to lookup functions, which always get a static placeholder
(i.e. MISSING) for UNNEST, and due to this check, the deserialization failed.
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/y6fgzyx330omhkr40376knw8k4oczz3s].
> The stacktrace is
> {code:java}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> the output type strategy is "Missing"; as a result,
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
> !screenshot-1.png|width=675,height=295!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17692450#comment-17692450
]
Jane Chan edited comment on FLINK-31182 at 2/23/23 3:02 AM:
[~Sergey Nuyanzin] Yes, the check on the output type inference strategy was
introduced in FunctionCatalogTableOperator in FLINK-15487 for FLIP-65 a long
time ago.
I guess it worked well before FLIP-190 was introduced
because`LogicalUnnestRule` will infer the output type and convert UNNEST to
EXPLODE. (see [LogicalUnnestRule
L#99|https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRule.scala#L99]).
As a result, although the BuiltinFunctionDefinition for UNNEST always has a
"MISSING" type inference, it does not affect the execution.
However, after CompiledPlan is introduced, RexNodeJsonDeserializer relies on
the OperatorTable to lookup functions, which always get a static placeholder
(i.e. MISSING) for UNNEST, and due to this check, the deserialization failed.
was (Author: qingyue):
[~Sergey Nuyanzin] Yes, the check on the output type inference strategy was
introduced in FunctionCatalogTableOperator in FLINK-15487 for FLIP-65 a long
time ago. I guess it worked well before FLIP-190 was introduced. The
RexNodeJsonDeserializer relies on the OperatorTable to lookup functions, and
due to this check, the deserialization failed.
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/y6fgzyx330omhkr40376knw8k4oczz3s].
> The stacktrace is
> {code:java}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> the output type strategy is "Missing"; as a result,
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
> !screenshot-1.png|width=675,height=295!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31182:
--
Description:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/y6fgzyx330omhkr40376knw8k4oczz3s].
The stacktrace is
{code:java}
Unable to find source-code formatter for language: text. Available languages
are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc, perl,
php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1', the
output type strategy is "Missing"; as a result,
FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
!screenshot-1.png|width=675,height=295!
was:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:java}
Unable to find source-code formatter for language: text. Available languages
are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc, perl,
php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1', the
output type strategy is "Missing"; as a result,
FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
!screenshot-1.png|width=675,height=295!
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/y6fgzyx330omhkr40376knw8k4oczz3s].
> The stacktrace is
> {code:java}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
>
[jira] [Commented] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17692450#comment-17692450
]
Jane Chan commented on FLINK-31182:
---
[~Sergey Nuyanzin] Yes, the check on the output type inference strategy was
introduced in FunctionCatalogTableOperator in FLINK-15487 for FLIP-65 a long
time ago. I guess it worked well before FLIP-190 was introduced. The
RexNodeJsonDeserializer relies on the OperatorTable to lookup functions, and
due to this check, the deserialization failed.
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code:java}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> the output type strategy is "Missing"; as a result,
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
> !screenshot-1.png|width=675,height=295!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17692156#comment-17692156
]
Jane Chan commented on FLINK-31182:
---
Correct me if I'm wrong, but for all BultinFunctionDefinition with output type
strategy as TypeStrategies.MISSING, the deserialization will fail. By removing
the check, the execution of the deserialized plan is successful. However, I'm
not very sure about the purpose of the check. Can anyone shed some light?
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code:java}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> the output type strategy is "Missing"; as a result,
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
> !screenshot-1.png|width=675,height=295!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31182:
--
Description:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:java}
Unable to find source-code formatter for language: text. Available languages
are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc, perl,
php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1', the
output type strategy is "Missing"; as a result,
FunctionCatalogOperatorTable#convertToBridgingSqlFunction returns empty.
!screenshot-1.png|width=675,height=295!
was:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code}
Unable to find source-code formatter for language: text. Available languages
are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc, perl,
php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
FunctionCatalogOperatorTable#convertToBridgingSqlFunction cannot infer output
type strategy.
!screenshot-1.png|width=675,height=295!
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code:java}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
>
[jira] [Updated] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31182:
--
Summary: CompiledPlan cannot deserialize BridgingSqlFunction with
MissingTypeStrategy (was: CompiledPlan cannot deserialize BridgingSqlFunction
correctly)
> CompiledPlan cannot deserialize BridgingSqlFunction with MissingTypeStrategy
>
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction cannot infer output
> type strategy.
> !screenshot-1.png|width=675,height=295!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction correctly
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31182:
--
Description:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code}
Unable to find source-code formatter for language: text. Available languages
are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc, perl,
php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
FunctionCatalogOperatorTable#convertToBridgingSqlFunction cannot infer output
type strategy.
!screenshot-1.png|width=675,height=295!
was:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:text}
Caused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
FunctionCatalogOperatorTable#convertToBridgingSqlFunction cannot infer output
type strategy.
> CompiledPlan cannot deserialize BridgingSqlFunction correctly
> -
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code}
> Unable to find source-code formatter for language: text. Available languages
> are: actionscript, ada, applescript, bash, c, c#, c++, cpp, css, erlang, go,
> groovy, haskell, html, java, javascript, js, json, lua, none, nyan, objc,
> perl, php, python, r, rainbow, ruby, scala, sh, sql, swift, visualbasic, xml,
> yamlCaused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
>
[jira] [Updated] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction correctly
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31182:
--
Attachment: screenshot-1.png
> CompiledPlan cannot deserialize BridgingSqlFunction correctly
> -
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
> Attachments: screenshot-1.png
>
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code:text}
> Caused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction cannot infer output
> type strategy.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction correctly
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31182:
--
Description:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:text}
Caused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
FunctionCatalogOperatorTable#convertToBridgingSqlFunction cannot infer output
type strategy.
was:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:text}
Caused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause can be identified by enabling the DEBUG level for ModuleManager
{code:text}
12089 [main] DEBUG org.apache.flink.table.module.ModuleManager [] - Got
FunctionDefinition '$unnest_rows$1' from 'core' module.
12089 [main] DEBUG org.apache.flink.table.module.ModuleManager [] - Cannot find
FunctionDefinition 'unnest_rows' from any loaded modules.
{code}
> CompiledPlan cannot deserialize BridgingSqlFunction correctly
> -
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code:text}
> Caused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause is that although ModuleManager can resolve '$UNNEST_ROWS$1',
> FunctionCatalogOperatorTable#convertToBridgingSqlFunction cannot infer output
> type strategy.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction correctly
[
https://issues.apache.org/jira/browse/FLINK-31182?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31182:
--
Description:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:text}
Caused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
The root cause can be identified by enabling the DEBUG level for ModuleManager
{code:text}
12089 [main] DEBUG org.apache.flink.table.module.ModuleManager [] - Got
FunctionDefinition '$unnest_rows$1' from 'core' module.
12089 [main] DEBUG org.apache.flink.table.module.ModuleManager [] - Cannot find
FunctionDefinition 'unnest_rows' from any loaded modules.
{code}
was:
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:java}
Caused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
> CompiledPlan cannot deserialize BridgingSqlFunction correctly
> -
>
> Key: FLINK-31182
> URL: https://issues.apache.org/jira/browse/FLINK-31182
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.17.0, 1.18.0, 1.17.1
>Reporter: Jane Chan
>Priority: Major
>
> This issue is reported from the [user mail
> list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
> The stacktrace is
> {code:text}
> Caused by: org.apache.flink.table.api.TableException: Could not resolve
> internal system function '$UNNEST_ROWS$1'. This is a bug, please file an
> issue.
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
> at
> org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
> {code}
> The root cause can be identified by enabling the DEBUG level for ModuleManager
> {code:text}
> 12089 [main] DEBUG org.apache.flink.table.module.ModuleManager [] - Got
> FunctionDefinition '$unnest_rows$1' from 'core' module.
> 12089 [main] DEBUG org.apache.flink.table.module.ModuleManager [] - Cannot
> find FunctionDefinition 'unnest_rows' from any loaded modules.
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-31182) CompiledPlan cannot deserialize BridgingSqlFunction correctly
Jane Chan created FLINK-31182:
-
Summary: CompiledPlan cannot deserialize BridgingSqlFunction
correctly
Key: FLINK-31182
URL: https://issues.apache.org/jira/browse/FLINK-31182
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Affects Versions: 1.17.0, 1.18.0, 1.17.1
Reporter: Jane Chan
This issue is reported from the [user mail
list|https://lists.apache.org/thread/dglf8zgtt1yx3vrdytn0dlv7b3pw1nq6].
The stacktrace is
{code:java}
Caused by: org.apache.flink.table.api.TableException: Could not resolve
internal system function '$UNNEST_ROWS$1'. This is a bug, please file an issue.
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeInternalFunction(RexNodeJsonDeserializer.java:392)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeSqlOperator(RexNodeJsonDeserializer.java:337)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserializeCall(RexNodeJsonDeserializer.java:307)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:146)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:128)
at
org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonDeserializer.deserialize(RexNodeJsonDeserializer.java:115)
{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-31091) SQL interval related queries stop working via SQL client
[
https://issues.apache.org/jira/browse/FLINK-31091?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17691352#comment-17691352
]
Jane Chan edited comment on FLINK-31091 at 2/21/23 3:31 AM:
Thanks, [~Sergey Nuyanzin] and [~mapohl] for reporting this.
The reason is that
org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer does not
cope with `INTERVAL_YEAR_MONTH` and `INTERVAL_DAY_TIME`. cc [~fsk119]
was (Author: qingyue):
Thanks, [~Sergey Nuyanzin] and [~mapohl] for reporting this.
The reason is that
org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer does not
cope with `INTERVAL_YEAR_MONTH` and `INTERVAL_DAY_TIME`.
I'd like to fix this issue. cc [~fsk119]
> SQL interval related queries stop working via SQL client
>
>
> Key: FLINK-31091
> URL: https://issues.apache.org/jira/browse/FLINK-31091
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Client
>Affects Versions: 1.17.0
>Reporter: Sergey Nuyanzin
>Assignee: Sergey Nuyanzin
>Priority: Blocker
> Labels: pull-request-available
>
> I put blocker since it works in 1.16.x and stopped working in 1.17 after a
> certain commit
> Any interval related query run via SQL Client is failing with
> {noformat}
> [ERROR] Could not execute SQL statement. Reason:
> org.apache.flink.runtime.rest.util.RestClientException: [Internal server
> error. Could not map response to JSON.]
> at
> org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
> at
> org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
> at
> java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:966)
> at
> java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:940)
> at
> java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:456)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:750)
> {noformat}
> example of query
> {code:sql}
> SELECT INTERVAL '2' DAY;
> SELECT 1, INTERVAL '2' YEAR;
> {code}
> based on tests it stopped working after this commit
> https://issues.apache.org/jira/browse/FLINK-29945
> More traces from logs
> {noformat}
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonMappingException:
> Unable to serialize logical type 'INTERVAL MONTH NOT NULL'. Please check the
> documentation for supported types. (through reference chain: java.util.Coll
> ections$UnmodifiableRandomAccessList[1]->org.apache.flink.table.gateway.rest.serde.ColumnInfo["logicalType"])
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonMappingException.wrapWithPath(JsonMappingException.java:392)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonMappingException.wrapWithPath(JsonMappingException.java:351)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer.wrapAndThrow(StdSerializer.java:316)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.BeanSerializerBase.serializeFields(BeanSerializerBase.java:782)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.BeanSerializer.serialize(BeanSerializer.java:178)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.IndexedListSerializer.serializeContents(IndexedListSerializer.java:119)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.IndexedListSerializer.serialize(IndexedListSerializer.java:79)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.IndexedListSerializer.serialize(IndexedListSerializer.java:18)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider.defaultSerializeField(SerializerProvider.java:1166)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.table.gateway.rest.serde.ResultInfoSerializer.serialize(ResultInfoSerializer.java:82)
> ~[flink-sql-gateway-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
>
[jira] [Commented] (FLINK-31091) SQL interval related queries stop working via SQL client
[
https://issues.apache.org/jira/browse/FLINK-31091?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17691352#comment-17691352
]
Jane Chan commented on FLINK-31091:
---
Thanks, [~Sergey Nuyanzin] and [~mapohl] for reporting this.
The reason is that
org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer does not
cope with `INTERVAL_YEAR_MONTH` and `INTERVAL_DAY_TIME`.
I'd like to fix this issue. cc [~fsk119]
> SQL interval related queries stop working via SQL client
>
>
> Key: FLINK-31091
> URL: https://issues.apache.org/jira/browse/FLINK-31091
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Client
>Affects Versions: 1.17.0
>Reporter: Sergey Nuyanzin
>Assignee: Sergey Nuyanzin
>Priority: Blocker
> Labels: pull-request-available
>
> I put blocker since it works in 1.16.x and stopped working in 1.17 after a
> certain commit
> Any interval related query run via SQL Client is failing with
> {noformat}
> [ERROR] Could not execute SQL statement. Reason:
> org.apache.flink.runtime.rest.util.RestClientException: [Internal server
> error. Could not map response to JSON.]
> at
> org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
> at
> org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
> at
> java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:966)
> at
> java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:940)
> at
> java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:456)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:750)
> {noformat}
> example of query
> {code:sql}
> SELECT INTERVAL '2' DAY;
> SELECT 1, INTERVAL '2' YEAR;
> {code}
> based on tests it stopped working after this commit
> https://issues.apache.org/jira/browse/FLINK-29945
> More traces from logs
> {noformat}
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonMappingException:
> Unable to serialize logical type 'INTERVAL MONTH NOT NULL'. Please check the
> documentation for supported types. (through reference chain: java.util.Coll
> ections$UnmodifiableRandomAccessList[1]->org.apache.flink.table.gateway.rest.serde.ColumnInfo["logicalType"])
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonMappingException.wrapWithPath(JsonMappingException.java:392)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonMappingException.wrapWithPath(JsonMappingException.java:351)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer.wrapAndThrow(StdSerializer.java:316)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.BeanSerializerBase.serializeFields(BeanSerializerBase.java:782)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.BeanSerializer.serialize(BeanSerializer.java:178)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.IndexedListSerializer.serializeContents(IndexedListSerializer.java:119)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.IndexedListSerializer.serialize(IndexedListSerializer.java:79)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.IndexedListSerializer.serialize(IndexedListSerializer.java:18)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider.defaultSerializeField(SerializerProvider.java:1166)
> ~[flink-dist-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.table.gateway.rest.serde.ResultInfoSerializer.serialize(ResultInfoSerializer.java:82)
> ~[flink-sql-gateway-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.table.gateway.rest.serde.ResultInfoSerializer.serialize(ResultInfoSerializer.java:47)
> ~[flink-sql-gateway-1.18-SNAPSHOT.jar:1.18-SNAPSHOT]
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider.defaultSerializeField(SerializerProvider.java:1166)
>
[jira] [Commented] (FLINK-31159) Release Testing: Verify FLINK-30583 Provide the flame graph to the subtask level
[ https://issues.apache.org/jira/browse/FLINK-31159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17691333#comment-17691333 ] Jane Chan commented on FLINK-31159: --- Thanks [~liangtl] for creating this, I'd like to do this testing work. > Release Testing: Verify FLINK-30583 Provide the flame graph to the subtask > level > > > Key: FLINK-31159 > URL: https://issues.apache.org/jira/browse/FLINK-31159 > Project: Flink > Issue Type: Sub-task > Components: Runtime / REST, Runtime / Web Frontend >Affects Versions: 1.17.0 >Reporter: Hong Liang Teoh >Priority: Major > Fix For: 1.17.0 > > Attachments: Map_record_distribution.png, all_subtasks.png, > subtask0.png, subtask3.png > > > The issue aims to verify FLINK-30583. > Please verify: > # When below conditions are met: > ** Job has more than 1 parallel subtask > ** Some subtasks are busy, whilst others are not (e.g. some receive records, > some don't) > ** The FlameGraph accurately reflects the busy subtask -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-31013) Session window aggregation cannot trigger window using event time
[
https://issues.apache.org/jira/browse/FLINK-31013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17690177#comment-17690177
]
Jane Chan commented on FLINK-31013:
---
[~xzw0223] What's your point?
According to the table schema
||id||type||date||
|1|T1|2018-01-24 00:00:00.000|
|2|T1|2018-01-26 00:00:00.000|
|1|T2|2018-01-28 00:00:00.000|
|1|T2|2018-01-28 00:00:00.000|
And the query with a session window gap of 1 DAY, the first two windows should
be triggered.
{code:sql}
select `id`,
`type`,
COUNT(1) as event_cnt,
session_start(`date`, interval '1' DAY) as ss,
session_end(`date`, interval '1' DAY) as se
from events group by `id`, `type`, session(`date`, interval '1' DAY);
{code}
Expected Output
||id||type||event_cnt||ss||se||
|1|T1|1|2018-01-24 00:00:00.000|2018-01-25 00:00:00.000|
|2|T1|1|2018-01-26 00:00:00.000|2018-01-27 00:00:00.000|
Table Store Actual Output
No output
> Session window aggregation cannot trigger window using event time
> -
>
> Key: FLINK-31013
> URL: https://issues.apache.org/jira/browse/FLINK-31013
> Project: Flink
> Issue Type: Bug
> Components: Table Store
>Affects Versions: table-store-0.4.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: table-store-0.4.0
>
>
> {code:sql}
> -- test against Flink 1.16.0
> create catalog fscat with (
> 'type' = 'table-store',
> 'warehouse' = 'file:///tmp/fscat'
> );
> use catalog fscat;
> create table events (
> `id` int,
> `type` string,
> `date` TIMESTAMP(3),
> watermark for `date` AS `date`);
>
> insert into events
> values (1, 'T1', to_timestamp('2018-01-24', '-MM-dd')),
> (2, 'T1', to_timestamp('2018-01-26', '-MM-dd')),
> (1, 'T2', to_timestamp('2018-01-28', '-MM-dd')),
> (1, 'T2', to_timestamp('2018-01-28', '-MM-dd'));
> -- no output
> select `id`,
> `type`,
> COUNT(1) as event_cnt,
> session_start(`date`, interval '1' DAY) as ss,
> session_end(`date`, interval '1' DAY) as se
> from events group by `id`, `type`, session(`date`, interval '1' DAY);
> -- explain plan
> == Abstract Syntax Tree ==
> LogicalProject(id=[$0], type=[$1], event_cnt=[$3], ss=[SESSION_START($2)],
> se=[SESSION_END($2)])
> +- LogicalAggregate(group=[{0, 1, 2}], event_cnt=[COUNT()])
> +- LogicalProject(id=[$0], type=[$1], $f2=[$SESSION($2, 8640:INTERVAL
> DAY)])
> +- LogicalWatermarkAssigner(rowtime=[date], watermark=[$2])
> +- LogicalTableScan(table=[[fscat, default, events]])
> == Optimized Physical Plan ==
> Calc(select=[id, type, event_cnt, w$start AS ss, w$end AS se])
> +- GroupWindowAggregate(groupBy=[id, type], window=[SessionGroupWindow('w$,
> date, 8640)], properties=[w$start, w$end, w$rowtime, w$proctime],
> select=[id, type, COUNT(*) AS event_cnt, start('w$) AS w$start, end('w$) AS
> w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
> +- Exchange(distribution=[hash[id, type]])
> +- TableSourceScan(table=[[fscat, default, events, watermark=[date]]],
> fields=[id, type, date])
> == Optimized Execution Plan ==
> Calc(select=[id, type, event_cnt, w$start AS ss, w$end AS se])
> +- GroupWindowAggregate(groupBy=[id, type], window=[SessionGroupWindow('w$,
> date, 8640)], properties=[w$start, w$end, w$rowtime, w$proctime],
> select=[id, type, COUNT(*) AS event_cnt, start('w$) AS w$start, end('w$) AS
> w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
> +- Exchange(distribution=[hash[id, type]])
> +- TableSourceScan(table=[[fscat, default, events, watermark=[date]]],
> fields=[id, type, date])
> -- however, if switch to filesystem source, the window can be triggered
> normally
> CREATE TEMPORARY TABLE `fscat`.`default`.`event_file_source` (
> `id` INT,
> `type` VARCHAR(2147483647),
> `date` TIMESTAMP(3),
> WATERMARK FOR `date` AS `date`
> ) WITH (
> 'format' = 'csv',
> 'path' = '/tmp/events.csv',
> 'source.monitor-interval' = '1 min',
> 'connector' = 'filesystem'
> );
> // cat events.csv
> 1,T1,2018-01-24 00:00:00.000
> 2,T1,2018-01-26 00:00:00.000
> 1,T2,2018-01-28 00:00:00.000
> 1,T2,2018-01-28 00:00:00.000
> -- same query using filesystem source
> select `id`, `type`, COUNT(1) as event_cnt, session_start(`date`, interval
> '1' DAY) as ss, session_end(`date`, interval '1' DAY) as se from
> event_file_source group by `id`, `type`, session(`date`, interval '1' DAY);
> -- output
> id type event_cnt
> ss se
> 1 T1 1 2018-01-24
> 00:00:00.000 2018-01-25 00:00:00.000
> 2 T1 1 2018-01-26
> 00:00:00.000 2018-01-27
[jira] [Commented] (FLINK-31025) Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide SQL advice
[
https://issues.apache.org/jira/browse/FLINK-31025?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17688845#comment-17688845
]
Jane Chan commented on FLINK-31025:
---
[~Weijie Guo] Thanks for testing thoroughly and finding the typo! I'll take a
look right now.
> Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide
> SQL advice
> ---
>
> Key: FLINK-31025
> URL: https://issues.apache.org/jira/browse/FLINK-31025
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
>Affects Versions: 1.17.0
>Reporter: Jane Chan
>Assignee: Weijie Guo
>Priority: Blocker
> Fix For: 1.17.0
>
> Attachments: image-2023-02-13-14-48-53-758.png,
> image-2023-02-13-14-52-41-855.png, test1.png
>
>
> This ticket aims for verifying FLINK-30650: Introduce EXPLAIN PLAN_ADVICE to
> provide SQL advice.
> More details about this feature and how to use it can be found in this
> [documentation|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/explain/].
> The verification is divided into two parts:
> Part I: Verify "EXPLAIN PLAN_ADVICE" can work with different queries under
> the streaming mode, such as a single select/insert/statement set w/ or w/o
> sub-plan reuse (configured by "table.optimizer.reuse-sub-plan-enabled").
> This indicates once specifying the ExplainDetail as "PLAN_ADVICE"
> {code:sql}
> EXPLAIN PLAN_ADVICE [SELECT ... FROM ...| INSERT INTO ...| EXECUTE STATEMENT
> SET BEGIN INSERT INTO ... END]
> {code}
> You should find the output should be like the following format (note that the
> title is changed to "Optimized Physical Plan with Advice")
> {code:sql}
> == Abstract Syntax Tree ==
> ...
>
> == Optimized Physical Plan With Advice ==
> ...
>
>
> == Optimized Execution Plan ==
> ...
> {code}
> The available advice is attached at the end of "== Optimized Physical Plan
> With Advice ==", and in front of "== Optimized Execution Plan =="
> If switching to batch mode, you should find the "EXPLAIN PLAN_ADVICE" should
> throw UnsupportedOperationException as expected.
>
> Part II: Verify the advice content
> Write a group aggregate query, and enable/disable the local-global two-phase
> configuration, and test the output.
> You should find once the following configurations are enabled, you will get
> the "no available advice..." output.
> {code:java}
> table.exec.mini-batch.enabled
> table.exec.mini-batch.allow-latency
> table.exec.mini-batch.size
> table.optimizer.agg-phase-strategy {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-31025) Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide SQL advice
[
https://issues.apache.org/jira/browse/FLINK-31025?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31025:
--
Description:
This ticket aims for verifying FLINK-30650: Introduce EXPLAIN PLAN_ADVICE to
provide SQL advice.
More details about this feature and how to use it can be found in this
[documentation|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/explain/].
The verification is divided into two parts:
Part I: Verify "EXPLAIN PLAN_ADVICE" can work with different queries under the
streaming mode, such as a single select/insert/statement set w/ or w/o sub-plan
reuse (configured by "table.optimizer.reuse-sub-plan-enabled").
This indicates once specifying the ExplainDetail as "PLAN_ADVICE"
{code:sql}
EXPLAIN PLAN_ADVICE [SELECT ... FROM ...| INSERT INTO ...| EXECUTE STATEMENT
SET BEGIN INSERT INTO ... END]
{code}
You should find the output should be like the following format (note that the
title is changed to "Optimized Physical Plan with Advice")
{code:sql}
== Abstract Syntax Tree ==
...
== Optimized Physical Plan With Advice ==
...
== Optimized Execution Plan ==
...
{code}
The available advice is attached at the end of "== Optimized Physical Plan With
Advice ==", and in front of "== Optimized Execution Plan =="
If switching to batch mode, you should find the "EXPLAIN PLAN_ADVICE" should
throw UnsupportedOperationException as expected.
Part II: Verify the advice content
Write a group aggregate query, and enable/disable the local-global two-phase
configuration, and test the output.
You should find once the following configurations are enabled, you will get the
"no available advice..." output.
{code:java}
table.exec.mini-batch.enabled
table.exec.mini-batch.allow-latency
table.exec.mini-batch.size
table.optimizer.agg-phase-strategy {code}
was:
This ticket aims for verifying FLINK-30650: Adaptive Batch Scheduler should
also work with hybrid shuffle mode.
More details about this feature and how to use it can be found in this
[documentation|xxx].
The verification is divided into two parts:
Part I: Verify hybrid shuffle can work with AdaptiveBatchScheduler
Write a simple Flink batch job using hybrid shuffle mode and submit this job.
Note that in flink-1.17, AdaptiveBatchScheduler is the default scheduler for
batch job, so you do not need other configuration.
Suppose your job's topology like source -> map -> sink, if your cluster have
enough slots, you should find that source and map are running at the same time.
Part II: Verify hybrid shuffle can work with Speculative Execution
Write a Flink batch job using hybrid shuffle mode which has a subtask running
much slower than others (e.g. sleep indefinitely if it runs on a certain host,
the hostname can be retrieved via InetAddress.getLocalHost().getHostName(), or
if its (subtaskIndex + attemptNumer) % 2 == 0)
Modify Flink configuration file to enable speculative execution and tune the
configuration as you like
Submit the job. Checking the web UI, logs, metrics and produced result.
You should find that once a producer task's one subtask finished, all its
consumer tasks can be scheduled in log.
> Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide
> SQL advice
> ---
>
> Key: FLINK-31025
> URL: https://issues.apache.org/jira/browse/FLINK-31025
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
>Affects Versions: 1.17.0
>Reporter: Jane Chan
>Priority: Blocker
> Fix For: 1.17.0
>
>
> This ticket aims for verifying FLINK-30650: Introduce EXPLAIN PLAN_ADVICE to
> provide SQL advice.
> More details about this feature and how to use it can be found in this
> [documentation|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/explain/].
> The verification is divided into two parts:
> Part I: Verify "EXPLAIN PLAN_ADVICE" can work with different queries under
> the streaming mode, such as a single select/insert/statement set w/ or w/o
> sub-plan reuse (configured by "table.optimizer.reuse-sub-plan-enabled").
> This indicates once specifying the ExplainDetail as "PLAN_ADVICE"
> {code:sql}
> EXPLAIN PLAN_ADVICE [SELECT ... FROM ...| INSERT INTO ...| EXECUTE STATEMENT
> SET BEGIN INSERT INTO ... END]
> {code}
> You should find the output should be like the following format (note that the
> title is changed to "Optimized Physical Plan with Advice")
> {code:sql}
> == Abstract Syntax Tree ==
> ...
>
> == Optimized Physical Plan With Advice ==
> ...
>
>
> == Optimized Execution Plan ==
> ...
> {code}
> The available advice is attached at the end of "== Optimized Physical Plan
> With Advice ==", and in front of "== Optimized Execution Plan =="
> If switching to
[jira] [Updated] (FLINK-31025) Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide SQL advice
[ https://issues.apache.org/jira/browse/FLINK-31025?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-31025: -- Description: This ticket aims for verifying FLINK-30650: Adaptive Batch Scheduler should also work with hybrid shuffle mode. More details about this feature and how to use it can be found in this [documentation|xxx]. The verification is divided into two parts: Part I: Verify hybrid shuffle can work with AdaptiveBatchScheduler Write a simple Flink batch job using hybrid shuffle mode and submit this job. Note that in flink-1.17, AdaptiveBatchScheduler is the default scheduler for batch job, so you do not need other configuration. Suppose your job's topology like source -> map -> sink, if your cluster have enough slots, you should find that source and map are running at the same time. Part II: Verify hybrid shuffle can work with Speculative Execution Write a Flink batch job using hybrid shuffle mode which has a subtask running much slower than others (e.g. sleep indefinitely if it runs on a certain host, the hostname can be retrieved via InetAddress.getLocalHost().getHostName(), or if its (subtaskIndex + attemptNumer) % 2 == 0) Modify Flink configuration file to enable speculative execution and tune the configuration as you like Submit the job. Checking the web UI, logs, metrics and produced result. You should find that once a producer task's one subtask finished, all its consumer tasks can be scheduled in log. > Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide > SQL advice > --- > > Key: FLINK-31025 > URL: https://issues.apache.org/jira/browse/FLINK-31025 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / API >Affects Versions: 1.17.0 >Reporter: Jane Chan >Priority: Blocker > Fix For: 1.17.0 > > > This ticket aims for verifying FLINK-30650: Adaptive Batch Scheduler should > also work with hybrid shuffle mode. > More details about this feature and how to use it can be found in this > [documentation|xxx]. > The verification is divided into two parts: > Part I: Verify hybrid shuffle can work with AdaptiveBatchScheduler > Write a simple Flink batch job using hybrid shuffle mode and submit this job. > Note that in flink-1.17, AdaptiveBatchScheduler is the default scheduler for > batch job, so you do not need other configuration. > Suppose your job's topology like source -> map -> sink, if your cluster have > enough slots, you should find that source and map are running at the same > time. > Part II: Verify hybrid shuffle can work with Speculative Execution > Write a Flink batch job using hybrid shuffle mode which has a subtask running > much slower than others (e.g. sleep indefinitely if it runs on a certain > host, the hostname can be retrieved via > InetAddress.getLocalHost().getHostName(), or if its (subtaskIndex + > attemptNumer) % 2 == 0) > Modify Flink configuration file to enable speculative execution and tune the > configuration as you like > Submit the job. Checking the web UI, logs, metrics and produced result. > You should find that once a producer task's one subtask finished, all its > consumer tasks can be scheduled in log. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-31025) Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide SQL advice
Jane Chan created FLINK-31025: - Summary: Release Testing: Verify FLINK-30650 Introduce EXPLAIN PLAN_ADVICE to provide SQL advice Key: FLINK-31025 URL: https://issues.apache.org/jira/browse/FLINK-31025 Project: Flink Issue Type: Sub-task Components: Table SQL / API Affects Versions: 1.17.0 Reporter: Jane Chan Fix For: 1.17.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-31013) Session window aggregation cannot trigger window using event time
[
https://issues.apache.org/jira/browse/FLINK-31013?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31013:
--
Description:
{code:java}
-- test against Flink 1.16.0
create catalog fscat with (
'type' = 'table-store',
'warehouse' = 'file:///tmp/fscat'
);
use catalog fscat;
create table events (
`id` int,
`type` string,
`date` TIMESTAMP(3),
watermark for `date` AS `date`);
insert into events
values (1, 'T1', to_timestamp('2018-01-24', '-MM-dd')),
(2, 'T1', to_timestamp('2018-01-26', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd'));
-- no output
select `id`,
`type`,
COUNT(1) as event_cnt,
session_start(`date`, interval '1' DAY) as ss,
session_end(`date`, interval '1' DAY) as se
from events group by `id`, `type`, session(`date`, interval '1' DAY);
-- explain plan
== Abstract Syntax Tree ==
LogicalProject(id=[$0], type=[$1], event_cnt=[$3], ss=[SESSION_START($2)],
se=[SESSION_END($2)])
+- LogicalAggregate(group=[{0, 1, 2}], event_cnt=[COUNT()])
+- LogicalProject(id=[$0], type=[$1], $f2=[$SESSION($2, 8640:INTERVAL
DAY)])
+- LogicalWatermarkAssigner(rowtime=[date], watermark=[$2])
+- LogicalTableScan(table=[[fscat, default, events]])
== Optimized Physical Plan ==
Calc(select=[id, type, event_cnt, w$start AS ss, w$end AS se])
+- GroupWindowAggregate(groupBy=[id, type], window=[SessionGroupWindow('w$,
date, 8640)], properties=[w$start, w$end, w$rowtime, w$proctime],
select=[id, type, COUNT(*) AS event_cnt, start('w$) AS w$start, end('w$) AS
w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
+- Exchange(distribution=[hash[id, type]])
+- TableSourceScan(table=[[fscat, default, events, watermark=[date]]],
fields=[id, type, date])
== Optimized Execution Plan ==
Calc(select=[id, type, event_cnt, w$start AS ss, w$end AS se])
+- GroupWindowAggregate(groupBy=[id, type], window=[SessionGroupWindow('w$,
date, 8640)], properties=[w$start, w$end, w$rowtime, w$proctime],
select=[id, type, COUNT(*) AS event_cnt, start('w$) AS w$start, end('w$) AS
w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
+- Exchange(distribution=[hash[id, type]])
+- TableSourceScan(table=[[fscat, default, events, watermark=[date]]],
fields=[id, type, date])
-- however, if switch to filesystem source, the window can be triggered normally
CREATE TEMPORARY TABLE `fscat`.`default`.`event_file_source` (
`id` INT,
`type` VARCHAR(2147483647),
`date` TIMESTAMP(3),
WATERMARK FOR `date` AS `date`
) WITH (
'format' = 'csv',
'path' = '/tmp/events.csv',
'source.monitor-interval' = '1 min',
'connector' = 'filesystem'
);
// cat events.csv
1,T1,2018-01-24 00:00:00.000
2,T1,2018-01-26 00:00:00.000
1,T2,2018-01-28 00:00:00.000
1,T2,2018-01-28 00:00:00.000
-- same query using filesystem source
select `id`, `type`, COUNT(1) as event_cnt, session_start(`date`, interval '1'
DAY) as ss, session_end(`date`, interval '1' DAY) as se from event_file_source
group by `id`, `type`, session(`date`, interval '1' DAY);
-- output
id type event_cnt
ss se
1 T1 1 2018-01-24
00:00:00.000 2018-01-25 00:00:00.000
2 T1 1 2018-01-26
00:00:00.000 2018-01-27 00:00:00.000{code}
was:
{code:java}
-- test against Flink 1.16.0
create catalog fscat with (
'type' = 'table-store',
'warehouse' = 'file:///tmp/fscat'
);
use catalog fscat;
create table events (
`id` int,
`type` string,
`date` TIMESTAMP(3),
watermark for `date` AS `date`);
insert into events
values (1, 'T1', to_timestamp('2018-01-24', '-MM-dd')),
(2, 'T1', to_timestamp('2018-01-26', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd'));
-- no output
select `id`,
`type`,
COUNT(1) as event_cnt,
session_start(`date`, interval '1' DAY) as ss,
session_end(`date`, interval '1' DAY) as se
from events group by `id`, `type`, session(`date`, interval '1' DAY); {code}
> Session window aggregation cannot trigger window using event time
> -
>
> Key: FLINK-31013
> URL: https://issues.apache.org/jira/browse/FLINK-31013
> Project: Flink
> Issue Type: Bug
> Components: Table Store
>Affects Versions: table-store-0.4.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: table-store-0.4.0
>
>
> {code:java}
> -- test against Flink 1.16.0
> create catalog fscat with (
> 'type' = 'table-store',
> 'warehouse' =
[jira] [Updated] (FLINK-31013) Session window aggregation cannot trigger window using event time
[
https://issues.apache.org/jira/browse/FLINK-31013?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31013:
--
Description:
{code:sql}
-- test against Flink 1.16.0
create catalog fscat with (
'type' = 'table-store',
'warehouse' = 'file:///tmp/fscat'
);
use catalog fscat;
create table events (
`id` int,
`type` string,
`date` TIMESTAMP(3),
watermark for `date` AS `date`);
insert into events
values (1, 'T1', to_timestamp('2018-01-24', '-MM-dd')),
(2, 'T1', to_timestamp('2018-01-26', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd'));
-- no output
select `id`,
`type`,
COUNT(1) as event_cnt,
session_start(`date`, interval '1' DAY) as ss,
session_end(`date`, interval '1' DAY) as se
from events group by `id`, `type`, session(`date`, interval '1' DAY);
-- explain plan
== Abstract Syntax Tree ==
LogicalProject(id=[$0], type=[$1], event_cnt=[$3], ss=[SESSION_START($2)],
se=[SESSION_END($2)])
+- LogicalAggregate(group=[{0, 1, 2}], event_cnt=[COUNT()])
+- LogicalProject(id=[$0], type=[$1], $f2=[$SESSION($2, 8640:INTERVAL
DAY)])
+- LogicalWatermarkAssigner(rowtime=[date], watermark=[$2])
+- LogicalTableScan(table=[[fscat, default, events]])
== Optimized Physical Plan ==
Calc(select=[id, type, event_cnt, w$start AS ss, w$end AS se])
+- GroupWindowAggregate(groupBy=[id, type], window=[SessionGroupWindow('w$,
date, 8640)], properties=[w$start, w$end, w$rowtime, w$proctime],
select=[id, type, COUNT(*) AS event_cnt, start('w$) AS w$start, end('w$) AS
w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
+- Exchange(distribution=[hash[id, type]])
+- TableSourceScan(table=[[fscat, default, events, watermark=[date]]],
fields=[id, type, date])
== Optimized Execution Plan ==
Calc(select=[id, type, event_cnt, w$start AS ss, w$end AS se])
+- GroupWindowAggregate(groupBy=[id, type], window=[SessionGroupWindow('w$,
date, 8640)], properties=[w$start, w$end, w$rowtime, w$proctime],
select=[id, type, COUNT(*) AS event_cnt, start('w$) AS w$start, end('w$) AS
w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
+- Exchange(distribution=[hash[id, type]])
+- TableSourceScan(table=[[fscat, default, events, watermark=[date]]],
fields=[id, type, date])
-- however, if switch to filesystem source, the window can be triggered normally
CREATE TEMPORARY TABLE `fscat`.`default`.`event_file_source` (
`id` INT,
`type` VARCHAR(2147483647),
`date` TIMESTAMP(3),
WATERMARK FOR `date` AS `date`
) WITH (
'format' = 'csv',
'path' = '/tmp/events.csv',
'source.monitor-interval' = '1 min',
'connector' = 'filesystem'
);
// cat events.csv
1,T1,2018-01-24 00:00:00.000
2,T1,2018-01-26 00:00:00.000
1,T2,2018-01-28 00:00:00.000
1,T2,2018-01-28 00:00:00.000
-- same query using filesystem source
select `id`, `type`, COUNT(1) as event_cnt, session_start(`date`, interval '1'
DAY) as ss, session_end(`date`, interval '1' DAY) as se from event_file_source
group by `id`, `type`, session(`date`, interval '1' DAY);
-- output
id type event_cnt
ss se
1 T1 1 2018-01-24
00:00:00.000 2018-01-25 00:00:00.000
2 T1 1 2018-01-26
00:00:00.000 2018-01-27 00:00:00.000{code}
was:
{code:java}
-- test against Flink 1.16.0
create catalog fscat with (
'type' = 'table-store',
'warehouse' = 'file:///tmp/fscat'
);
use catalog fscat;
create table events (
`id` int,
`type` string,
`date` TIMESTAMP(3),
watermark for `date` AS `date`);
insert into events
values (1, 'T1', to_timestamp('2018-01-24', '-MM-dd')),
(2, 'T1', to_timestamp('2018-01-26', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd'));
-- no output
select `id`,
`type`,
COUNT(1) as event_cnt,
session_start(`date`, interval '1' DAY) as ss,
session_end(`date`, interval '1' DAY) as se
from events group by `id`, `type`, session(`date`, interval '1' DAY);
-- explain plan
== Abstract Syntax Tree ==
LogicalProject(id=[$0], type=[$1], event_cnt=[$3], ss=[SESSION_START($2)],
se=[SESSION_END($2)])
+- LogicalAggregate(group=[{0, 1, 2}], event_cnt=[COUNT()])
+- LogicalProject(id=[$0], type=[$1], $f2=[$SESSION($2, 8640:INTERVAL
DAY)])
+- LogicalWatermarkAssigner(rowtime=[date], watermark=[$2])
+- LogicalTableScan(table=[[fscat, default, events]])
== Optimized Physical Plan ==
Calc(select=[id, type, event_cnt, w$start AS ss, w$end AS se])
+- GroupWindowAggregate(groupBy=[id, type], window=[SessionGroupWindow('w$,
date, 8640)],
[jira] [Updated] (FLINK-31013) Session window aggregation cannot trigger window using event time
[
https://issues.apache.org/jira/browse/FLINK-31013?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jane Chan updated FLINK-31013:
--
Summary: Session window aggregation cannot trigger window using event time
(was: SessionWindowAggregation cannot get trigger window using event time)
> Session window aggregation cannot trigger window using event time
> -
>
> Key: FLINK-31013
> URL: https://issues.apache.org/jira/browse/FLINK-31013
> Project: Flink
> Issue Type: Bug
> Components: Table Store
>Affects Versions: table-store-0.4.0
>Reporter: Jane Chan
>Priority: Major
> Fix For: table-store-0.4.0
>
>
> {code:java}
> -- test against Flink 1.16.0
> create catalog fscat with (
> 'type' = 'table-store',
> 'warehouse' = 'file:///tmp/fscat'
> );
> use catalog fscat;
> create table events (
> `id` int,
> `type` string,
> `date` TIMESTAMP(3),
> watermark for `date` AS `date`);
>
> insert into events
> values (1, 'T1', to_timestamp('2018-01-24', '-MM-dd')),
> (2, 'T1', to_timestamp('2018-01-26', '-MM-dd')),
> (1, 'T2', to_timestamp('2018-01-28', '-MM-dd')),
> (1, 'T2', to_timestamp('2018-01-28', '-MM-dd'));
> -- no output
> select `id`,
> `type`,
> COUNT(1) as event_cnt,
> session_start(`date`, interval '1' DAY) as ss,
> session_end(`date`, interval '1' DAY) as se
> from events group by `id`, `type`, session(`date`, interval '1' DAY); {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-31013) SessionWindowAggregation cannot get trigger window using event time
Jane Chan created FLINK-31013:
-
Summary: SessionWindowAggregation cannot get trigger window using
event time
Key: FLINK-31013
URL: https://issues.apache.org/jira/browse/FLINK-31013
Project: Flink
Issue Type: Bug
Components: Table Store
Affects Versions: table-store-0.4.0
Reporter: Jane Chan
Fix For: table-store-0.4.0
{code:java}
-- test against Flink 1.16.0
create catalog fscat with (
'type' = 'table-store',
'warehouse' = 'file:///tmp/fscat'
);
use catalog fscat;
create table events (
`id` int,
`type` string,
`date` TIMESTAMP(3),
watermark for `date` AS `date`);
insert into events
values (1, 'T1', to_timestamp('2018-01-24', '-MM-dd')),
(2, 'T1', to_timestamp('2018-01-26', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd')),
(1, 'T2', to_timestamp('2018-01-28', '-MM-dd'));
-- no output
select `id`,
`type`,
COUNT(1) as event_cnt,
session_start(`date`, interval '1' DAY) as ss,
session_end(`date`, interval '1' DAY) as se
from events group by `id`, `type`, session(`date`, interval '1' DAY); {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (FLINK-30958) Rest API doc generation failure caused by JobClientHeartbeatHeaders
Jane Chan created FLINK-30958: - Summary: Rest API doc generation failure caused by JobClientHeartbeatHeaders Key: FLINK-30958 URL: https://issues.apache.org/jira/browse/FLINK-30958 Project: Flink Issue Type: Bug Components: Runtime / REST Affects Versions: 1.17.0 Reporter: Jane Chan Fix For: 1.17.0 `JobClientHeartbeatHeaders` should override `operationId` since `getHttpMethod` returns `POST`. Otherwise `UnsupportedOperationException` is thrown at `OpenApiSpecGenerator` when generating the REST API doc. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-29405) InputFormatCacheLoaderTest is unstable
[
https://issues.apache.org/jira/browse/FLINK-29405?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17684385#comment-17684385
]
Jane Chan commented on FLINK-29405:
---
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=45662=logs=086353db-23b2-5446-2315-18e660618ef2=6cd785f3-2a2e-58a8-8e69-b4a03be28843=10314
> InputFormatCacheLoaderTest is unstable
> --
>
> Key: FLINK-29405
> URL: https://issues.apache.org/jira/browse/FLINK-29405
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.16.0, 1.17.0
>Reporter: Chesnay Schepler
>Assignee: Qingsheng Ren
>Priority: Blocker
> Labels: pull-request-available, test-stability
> Fix For: 1.17.0, 1.16.2
>
>
> #testExceptionDuringReload/#testCloseAndInterruptDuringReload fail reliably
> when run in a loop.
> {code}
> java.lang.AssertionError:
> Expecting AtomicInteger(0) to have value:
> 0
> but did not.
> at
> org.apache.flink.table.runtime.functions.table.fullcache.inputformat.InputFormatCacheLoaderTest.testCloseAndInterruptDuringReload(InputFormatCacheLoaderTest.java:161)
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-30811) Fix sql gateway can not stop job correctly
[ https://issues.apache.org/jira/browse/FLINK-30811?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17683792#comment-17683792 ] Jane Chan commented on FLINK-30811: --- Hi [~Paul Lin], yes, the PR has been merged. This issue is discovered when reviewing FLINK-29945. You can reach the discussion at https://github.com/apache/flink/pull/21717#discussion_r1089640498. I've rebased your fix and added the test case to verify that all problems have been fixed. > Fix sql gateway can not stop job correctly > -- > > Key: FLINK-30811 > URL: https://issues.apache.org/jira/browse/FLINK-30811 > Project: Flink > Issue Type: Bug > Components: Table SQL / Gateway >Affects Versions: 1.17.0 >Reporter: Shengkai Fang >Assignee: Jane Chan >Priority: Major > Labels: pull-request-available > Fix For: 1.17.0 > > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-30848) Enable CatalogTableITCase#testAlterTable
[ https://issues.apache.org/jira/browse/FLINK-30848?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jane Chan updated FLINK-30848: -- Description: This test has been ignored since FLINK-22315 because, at that moment, the `CREATE TABLE` and `ALTER TABLE`statements are creating deprecated CatalogTableImpl and DefaultCatalogTable interchangeably, which causes CI failure [https://dev.azure.com/qingyuecqy/Flink/_build/results?buildId=102=logs=43a593e7-535d-554b-08cc-244368da36b4=82d122c0-8bbf-56f3-4c0d-8e3d69630d0f] While after resolving FLINK-29679, this test should be enabled again. > Enable CatalogTableITCase#testAlterTable > > > Key: FLINK-30848 > URL: https://issues.apache.org/jira/browse/FLINK-30848 > Project: Flink > Issue Type: Technical Debt > Components: Table SQL / API >Affects Versions: 1.17.0 >Reporter: Jane Chan >Priority: Major > > This test has been ignored since FLINK-22315 because, at that moment, the > `CREATE TABLE` and `ALTER TABLE`statements are creating deprecated > CatalogTableImpl and DefaultCatalogTable interchangeably, which causes CI > failure > [https://dev.azure.com/qingyuecqy/Flink/_build/results?buildId=102=logs=43a593e7-535d-554b-08cc-244368da36b4=82d122c0-8bbf-56f3-4c0d-8e3d69630d0f] > > While after resolving FLINK-29679, this test should be enabled again. -- This message was sent by Atlassian Jira (v8.20.10#820010)
