You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Shengkai Fang (Jira)" <ji...@apache.org> on 2022/05/23 07:43:00 UTC
[jira] [Commented] (FLINK-24735) SQL client crashes with `Cannot add expression of different type to set`
[ https://issues.apache.org/jira/browse/FLINK-24735?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17540811#comment-17540811 ]
Shengkai Fang commented on FLINK-24735:
---------------------------------------
The main cause is the sql-client doesn't catch the error thrown by the planner, which causes the sql client crash. We can just catch the root of the Error and Exception, that is, we catch the Throwable in the Sql client.
> SQL client crashes with `Cannot add expression of different type to set`
> ------------------------------------------------------------------------
>
> Key: FLINK-24735
> URL: https://issues.apache.org/jira/browse/FLINK-24735
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Client
> Affects Versions: 1.14.0
> Reporter: Martijn Visser
> Assignee: Shengkai Fang
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.16.0, 1.14.5
>
>
> Reproductions steps:
> 1. Download airports.csv from https://www.kaggle.com/usdot/flight-delays
> 2. Start Flink SQL client and create table
> {code:sql}
> CREATE TABLE `airports` (
> `IATA_CODE` CHAR(3),
> `AIRPORT` STRING,
> `CITY` STRING,
> `STATE` CHAR(2),
> `COUNTRY` CHAR(3),
> `LATITUDE` DOUBLE NULL,
> `LONGITUDE` DOUBLE NULL,
> PRIMARY KEY (`IATA_CODE`) NOT ENFORCED
> ) WITH (
> 'connector' = 'filesystem',
> 'path' = 'file:///flink-sql-cookbook/other-builtin-functions/04_override_table_options/airports.csv',
> 'format' = 'csv'
> );
> {code}
> 3. Run the following SQL statement:
> {code:sql}
> SELECT * FROM `airports` /*+ OPTIONS('csv.ignore-parse-errors'='true') */ WHERE COALESCE(`IATA_CODE`, `AIRPORT`) IS NULL;
> {code}
> Stacktrace:
> {code:bash}
> Exception in thread "main" org.apache.flink.table.client.SqlClientException: Unexpected exception. This is a bug. Please consider filing an issue.
> at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:201)
> at org.apache.flink.table.client.SqlClient.main(SqlClient.java:161)
> Caused by: java.lang.AssertionError: Cannot add expression of different type to set:
> set type is RecordType(CHAR(3) CHARACTER SET "UTF-16LE" NOT NULL IATA_CODE, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" AIRPORT, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" CITY, CHAR(2) CHARACTER SET "UTF-16LE" STATE, CHAR(3) CHARACTER SET "UTF-16LE" COUNTRY, DOUBLE LATITUDE, DOUBLE LONGITUDE) NOT NULL
> expression type is RecordType(CHAR(3) CHARACTER SET "UTF-16LE" IATA_CODE, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" AIRPORT, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" CITY, CHAR(2) CHARACTER SET "UTF-16LE" STATE, CHAR(3) CHARACTER SET "UTF-16LE" COUNTRY, DOUBLE LATITUDE, DOUBLE LONGITUDE) NOT NULL
> set is rel#426:LogicalProject.NONE.any.None: 0.[NONE].[NONE](input=HepRelVertex#425,inputs=0..6)
> expression is LogicalProject(IATA_CODE=[null:CHAR(3) CHARACTER SET "UTF-16LE"], AIRPORT=[$1], CITY=[$2], STATE=[$3], COUNTRY=[$4], LATITUDE=[$5], LONGITUDE=[$6])
> LogicalFilter(condition=[IS NULL(CAST($0):VARCHAR(2147483647) CHARACTER SET "UTF-16LE")])
> LogicalTableScan(table=[[default_catalog, default_database, airports]], hints=[[[OPTIONS inheritPath:[] options:{csv.ignore-parse-errors=true}]]])
> at org.apache.calcite.plan.RelOptUtil.verifyTypeEquivalence(RelOptUtil.java:381)
> at org.apache.calcite.plan.hep.HepRuleCall.transformTo(HepRuleCall.java:58)
> at org.apache.calcite.plan.RelOptRuleCall.transformTo(RelOptRuleCall.java:268)
> at org.apache.calcite.plan.RelOptRuleCall.transformTo(RelOptRuleCall.java:283)
> at org.apache.calcite.rel.rules.ReduceExpressionsRule$ProjectReduceExpressionsRule.onMatch(ReduceExpressionsRule.java:310)
> at org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
> at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
> at org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
> at org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
> at org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
> at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
> at org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
> at org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
> at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
> at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
> at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
> at scala.collection.Iterator.foreach(Iterator.scala:937)
> at scala.collection.Iterator.foreach$(Iterator.scala:937)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
> at scala.collection.IterableLike.foreach(IterableLike.scala:70)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
> at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
> at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
> at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:81)
> at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:300)
> at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:183)
> at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1665)
> at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:805)
> at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1274)
> at org.apache.flink.table.client.gateway.local.LocalExecutor.lambda$executeOperation$3(LocalExecutor.java:209)
> at org.apache.flink.table.client.gateway.context.ExecutionContext.wrapClassLoader(ExecutionContext.java:88)
> at org.apache.flink.table.client.gateway.local.LocalExecutor.executeOperation(LocalExecutor.java:209)
> at org.apache.flink.table.client.gateway.local.LocalExecutor.executeQuery(LocalExecutor.java:231)
> at org.apache.flink.table.client.cli.CliClient.callSelect(CliClient.java:532)
> at org.apache.flink.table.client.cli.CliClient.callOperation(CliClient.java:423)
> at org.apache.flink.table.client.cli.CliClient.lambda$executeStatement$1(CliClient.java:332)
> at java.util.Optional.ifPresent(Optional.java:159)
> at org.apache.flink.table.client.cli.CliClient.executeStatement(CliClient.java:325)
> at org.apache.flink.table.client.cli.CliClient.executeInteractive(CliClient.java:297)
> at org.apache.flink.table.client.cli.CliClient.executeInInteractiveMode(CliClient.java:221)
> at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:151)
> at org.apache.flink.table.client.SqlClient.start(SqlClient.java:95)
> at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:187)
> ... 1 more
> {code}
> Any exception/error should be wrapped into a SqlExecutionException (in the executor) or SqlClientException in the highest layer, which is not happening in this case.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)