You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2016/08/17 14:47:11 UTC
flink git commit: [FLINK-4281] [table] Wrap all Calcite Exceptions in
Flink Exceptions
Repository: flink
Updated Branches:
refs/heads/master 9f928d1ec -> 7ce42c2e7
[FLINK-4281] [table] Wrap all Calcite Exceptions in Flink Exceptions
This closes #2372.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7ce42c2e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7ce42c2e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7ce42c2e
Branch: refs/heads/master
Commit: 7ce42c2e7e332b0fafd19a3f9ed49e4554958fdd
Parents: 9f928d1
Author: Jark Wu <wu...@alibaba-inc.com>
Authored: Mon Aug 15 23:25:08 2016 +0800
Committer: twalthr <tw...@apache.org>
Committed: Wed Aug 17 16:46:23 2016 +0200
----------------------------------------------------------------------
.../flink/api/table/FlinkPlannerImpl.scala | 56 +++++++++++---------
.../org/apache/flink/api/table/exceptions.scala | 12 +++++
2 files changed, 43 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7ce42c2e/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
index f677784..bb08654 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
@@ -29,7 +29,7 @@ import org.apache.calcite.rel.RelRoot
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rex.RexBuilder
import org.apache.calcite.schema.SchemaPlus
-import org.apache.calcite.sql.parser.{SqlParseException, SqlParser}
+import org.apache.calcite.sql.parser.{SqlParseException => CSqlParseException, SqlParser}
import org.apache.calcite.sql.validate.SqlValidator
import org.apache.calcite.sql.{SqlNode, SqlOperatorTable}
import org.apache.calcite.sql2rel.{RelDecorrelator, SqlRexConvertletTable, SqlToRelConverter}
@@ -55,9 +55,9 @@ class FlinkPlannerImpl(
val convertletTable: SqlRexConvertletTable = config.getConvertletTable
val defaultSchema: SchemaPlus = config.getDefaultSchema
- var validator: FlinkCalciteSqlValidator = null
- var validatedSqlNode: SqlNode = null
- var root: RelRoot = null
+ var validator: FlinkCalciteSqlValidator = _
+ var validatedSqlNode: SqlNode = _
+ var root: RelRoot = _
private def ready() {
if (this.traitDefs != null) {
@@ -68,15 +68,18 @@ class FlinkPlannerImpl(
}
}
- @throws(classOf[SqlParseException])
def parse(sql: String): SqlNode = {
- ready()
- val parser: SqlParser = SqlParser.create(sql, parserConfig)
- val sqlNode: SqlNode = parser.parseStmt
- sqlNode
+ try {
+ ready()
+ val parser: SqlParser = SqlParser.create(sql, parserConfig)
+ val sqlNode: SqlNode = parser.parseStmt
+ sqlNode
+ } catch {
+ case e: CSqlParseException =>
+ throw SqlParserException(s"SQL parse failed. ${e.getMessage}", e)
+ }
}
- @throws(classOf[ValidationException])
def validate(sqlNode: SqlNode): SqlNode = {
validator = new FlinkCalciteSqlValidator(operatorTable, createCatalogReader, typeFactory)
validator.setIdentifierExpansion(true)
@@ -85,24 +88,27 @@ class FlinkPlannerImpl(
}
catch {
case e: RuntimeException =>
- throw new ValidationException(s"SQL validation failed.", e)
+ throw new ValidationException(s"SQL validation failed. ${e.getMessage}", e)
}
validatedSqlNode
}
- @throws(classOf[RelConversionException])
def rel(sql: SqlNode): RelRoot = {
- assert(validatedSqlNode != null)
- val rexBuilder: RexBuilder = createRexBuilder
- val cluster: RelOptCluster = RelOptCluster.create(planner, rexBuilder)
- val sqlToRelConverter: SqlToRelConverter = new SqlToRelConverter(
- new ViewExpanderImpl, validator, createCatalogReader, cluster, convertletTable)
- sqlToRelConverter.setTrimUnusedFields(false)
- sqlToRelConverter.enableTableAccessConversion(false)
- root = sqlToRelConverter.convertQuery(validatedSqlNode, false, true)
- root = root.withRel(sqlToRelConverter.flattenTypes(root.rel, true))
- root = root.withRel(RelDecorrelator.decorrelateQuery(root.rel))
- root
+ try {
+ assert(validatedSqlNode != null)
+ val rexBuilder: RexBuilder = createRexBuilder
+ val cluster: RelOptCluster = RelOptCluster.create(planner, rexBuilder)
+ val sqlToRelConverter: SqlToRelConverter = new SqlToRelConverter(
+ new ViewExpanderImpl, validator, createCatalogReader, cluster, convertletTable)
+ sqlToRelConverter.setTrimUnusedFields(false)
+ sqlToRelConverter.enableTableAccessConversion(false)
+ root = sqlToRelConverter.convertQuery(validatedSqlNode, false, true)
+ root = root.withRel(sqlToRelConverter.flattenTypes(root.rel, true))
+ root = root.withRel(RelDecorrelator.decorrelateQuery(root.rel))
+ root
+ } catch {
+ case e: RelConversionException => throw TableException(e.getMessage)
+ }
}
/** Implements [[org.apache.calcite.plan.RelOptTable.ViewExpander]]
@@ -120,8 +126,8 @@ class FlinkPlannerImpl(
sqlNode = parser.parseQuery
}
catch {
- case e: SqlParseException =>
- throw new RuntimeException("parse failed", e)
+ case e: CSqlParseException =>
+ throw SqlParserException(s"SQL parse failed. ${e.getMessage}", e)
}
val catalogReader: CalciteCatalogReader = createCatalogReader.withSchemaPath(schemaPath)
val validator: SqlValidator =
http://git-wip-us.apache.org/repos/asf/flink/blob/7ce42c2e/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/exceptions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/exceptions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/exceptions.scala
index cd5ce6a..773e256 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/exceptions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/exceptions.scala
@@ -24,6 +24,18 @@ package org.apache.flink.api.table
case class ExpressionParserException(msg: String) extends RuntimeException(msg)
/**
+ * Exception for all errors occurring during sql parsing.
+ */
+case class SqlParserException(
+ msg: String,
+ cause: Throwable)
+ extends RuntimeException(msg, cause) {
+
+ def this(msg: String) = this(msg, null)
+
+}
+
+/**
* General Exception for all errors during table handling.
*/
case class TableException(msg: String) extends RuntimeException(msg)