Details
-
Bug
-
Status: Resolved
-
Major
-
Resolution: Fixed
-
1.14.0
Description
Reproductions steps:
1. Download airports.csv from https://www.kaggle.com/usdot/flight-delays
2. Start Flink SQL client and create table
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' );
3. Run the following SQL statement:
SELECT * FROM `airports` /*+ OPTIONS('csv.ignore-parse-errors'='true') */ WHERE COALESCE(`IATA_CODE`, `AIRPORT`) IS NULL;
Stacktrace:
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
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.