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 2018/10/30 10:26:12 UTC
[flink] 02/07: [FLINK-10687] [table] Move TableException and
ValidationException to flink-table-common
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f0b244e15720d698a486d778085cc5ab6ceb5f73
Author: Timo Walther <tw...@apache.org>
AuthorDate: Fri Oct 26 13:35:17 2018 +0200
[FLINK-10687] [table] Move TableException and ValidationException to flink-table-common
---
.../org/apache/flink/table/api/TableException.java | 39 ++++++++++++++++++++++
.../flink/table/api/ValidationException.java | 38 +++++++++++++++++++++
.../flink/table/api/BatchTableEnvironment.scala | 6 ++--
.../flink/table/api/StreamTableEnvironment.scala | 10 +++---
.../apache/flink/table/api/TableEnvironment.scala | 17 +++++-----
.../org/apache/flink/table/api/exceptions.scala | 37 --------------------
.../scala/org/apache/flink/table/api/table.scala | 16 ++++-----
.../flink/table/calcite/FlinkPlannerImpl.scala | 2 +-
.../flink/table/calcite/FlinkTypeFactory.scala | 13 ++++----
.../apache/flink/table/codegen/CodeGenerator.scala | 2 +-
.../table/descriptors/LiteralValueValidator.scala | 4 +--
.../flink/table/expressions/ExpressionUtils.scala | 2 +-
.../org/apache/flink/table/expressions/call.scala | 2 +-
.../flink/table/expressions/fieldExpression.scala | 2 +-
.../functions/utils/UserDefinedFunctionUtils.scala | 9 ++---
.../flink/table/plan/nodes/CommonCorrelate.scala | 2 +-
.../flink/table/plan/nodes/FlinkRelNode.scala | 2 +-
.../table/plan/nodes/dataset/DataSetJoin.scala | 10 +++---
.../table/plan/nodes/dataset/DataSetSort.scala | 2 +-
.../DataStreamGroupWindowAggregate.scala | 2 +-
.../plan/nodes/datastream/DataStreamJoin.scala | 2 +-
.../plan/nodes/datastream/DataStreamMatch.scala | 12 +++----
.../rules/common/LogicalWindowAggregateRule.scala | 6 ++--
.../plan/rules/common/WindowPropertiesRule.scala | 5 +--
.../rules/dataSet/DataSetWindowAggregateRule.scala | 4 +--
.../datastream/DataStreamGroupAggregateRule.scala | 2 +-
.../DataStreamGroupWindowAggregateRule.scala | 2 +-
.../DataStreamLogicalWindowAggregateRule.scala | 6 ++--
.../plan/rules/logical/LogicalUnnestRule.scala | 4 +--
.../table/runtime/aggregate/AggregateUtil.scala | 13 +++++---
.../flink/table/sources/CsvTableSource.scala | 2 +-
.../flink/table/sources/TableSourceUtil.scala | 27 +++++++--------
.../table/sources/tsextractors/ExistingField.scala | 2 +-
.../flink/table/validate/FunctionCatalog.scala | 14 ++++----
.../flink/table/descriptors/RowtimeTest.scala | 2 +-
35 files changed, 184 insertions(+), 136 deletions(-)
diff --git a/flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/api/TableException.java b/flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/api/TableException.java
new file mode 100644
index 0000000..caef3a8
--- /dev/null
+++ b/flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/api/TableException.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.api;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * General Exception for all errors during table handling.
+ *
+ * <p>This exception indicates that an internal error occurred or that a feature is not supported
+ * yet. Usually, this exception does not indicate a fault of the user.
+ */
+@PublicEvolving
+public class TableException extends RuntimeException {
+
+ public TableException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public TableException(String message) {
+ super(message);
+ }
+}
diff --git a/flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/api/ValidationException.java b/flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/api/ValidationException.java
new file mode 100644
index 0000000..c6ac1dc
--- /dev/null
+++ b/flink-libraries/flink-table-common/src/main/java/org/apache/flink/table/api/ValidationException.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.api;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Exception for all errors occurring during validation phase.
+ *
+ * <p>This exception indicates that the user did something wrong.
+ */
+@PublicEvolving
+public class ValidationException extends RuntimeException {
+
+ public ValidationException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public ValidationException(String message) {
+ super(message);
+ }
+}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
index 5a34ee1..6a7a921 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
@@ -205,8 +205,8 @@ abstract class BatchTableEnvironment(
tableSink: TableSink[_]): Unit = {
// validate
checkValidTableName(name)
- if (fieldNames == null) throw TableException("fieldNames must not be null.")
- if (fieldTypes == null) throw TableException("fieldTypes must not be null.")
+ if (fieldNames == null) throw new TableException("fieldNames must not be null.")
+ if (fieldTypes == null) throw new TableException("fieldTypes must not be null.")
if (fieldNames.length == 0) throw new TableException("fieldNames must not be empty.")
if (fieldNames.length != fieldTypes.length) {
throw new TableException("Same number of field names and types required.")
@@ -513,7 +513,7 @@ abstract class BatchTableEnvironment(
}
case _ =>
- throw TableException("Cannot generate DataSet due to an invalid logical plan. " +
+ throw new TableException("Cannot generate DataSet due to an invalid logical plan. " +
"This is a bug and should not happen. Please file an issue.")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
index 6519120..4973f34 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
@@ -120,7 +120,7 @@ abstract class StreamTableEnvironment(
// check that event-time is enabled if table source includes rowtime attributes
if (TableSourceUtil.hasRowtimeAttribute(streamTableSource) &&
execEnv.getStreamTimeCharacteristic != TimeCharacteristic.EventTime) {
- throw TableException(
+ throw new TableException(
s"A rowtime attribute requires an EventTime time characteristic in stream " +
s"environment. But is: ${execEnv.getStreamTimeCharacteristic}")
}
@@ -230,8 +230,8 @@ abstract class StreamTableEnvironment(
tableSink: TableSink[_]): Unit = {
checkValidTableName(name)
- if (fieldNames == null) throw TableException("fieldNames must not be null.")
- if (fieldTypes == null) throw TableException("fieldTypes must not be null.")
+ if (fieldNames == null) throw new TableException("fieldNames must not be null.")
+ if (fieldTypes == null) throw new TableException("fieldTypes must not be null.")
if (fieldNames.length == 0) throw new TableException("fieldNames must not be empty.")
if (fieldNames.length != fieldTypes.length) {
throw new TableException("Same number of field names and types required.")
@@ -550,7 +550,7 @@ abstract class StreamTableEnvironment(
// check if event-time is enabled
if (rowtime.isDefined && execEnv.getStreamTimeCharacteristic != TimeCharacteristic.EventTime) {
- throw TableException(
+ throw new TableException(
s"A rowtime attribute requires an EventTime time characteristic in stream environment. " +
s"But is: ${execEnv.getStreamTimeCharacteristic}")
}
@@ -966,7 +966,7 @@ abstract class StreamTableEnvironment(
case node: DataStreamRel =>
node.translateToPlan(this, queryConfig)
case _ =>
- throw TableException("Cannot generate DataStream due to an invalid logical plan. " +
+ throw new TableException("Cannot generate DataStream due to an invalid logical plan. " +
"This is a bug and should not happen. Please file an issue.")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index 7cfe249..3cc97d3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -356,7 +356,7 @@ abstract class TableEnvironment(val config: TableConfig) {
throw new TableException(
s"Cannot generate a valid execution plan for the given query: \n\n" +
s"${RelOptUtil.toString(input)}\n" +
- s"${t.msg}\n" +
+ s"${t.getMessage}\n" +
s"Please check the documentation for the set of currently supported SQL features.")
case a: AssertionError =>
// keep original exception stack for caller
@@ -817,13 +817,13 @@ abstract class TableEnvironment(val config: TableConfig) {
private[flink] def insertInto(table: Table, sinkTableName: String, conf: QueryConfig): Unit = {
// check that sink table exists
- if (null == sinkTableName) throw TableException("Name of TableSink must not be null.")
- if (sinkTableName.isEmpty) throw TableException("Name of TableSink must not be empty.")
+ if (null == sinkTableName) throw new TableException("Name of TableSink must not be null.")
+ if (sinkTableName.isEmpty) throw new TableException("Name of TableSink must not be empty.")
getTable(sinkTableName) match {
case None =>
- throw TableException(s"No table was registered under the name $sinkTableName.")
+ throw new TableException(s"No table was registered under the name $sinkTableName.")
case Some(s: TableSourceSinkTable[_, _]) if s.tableSinkTable.isDefined =>
val tableSink = s.tableSinkTable.get.tableSink
@@ -845,7 +845,7 @@ abstract class TableEnvironment(val config: TableConfig) {
.map { case (n, t) => s"$n: ${t.getTypeClass.getSimpleName}" }
.mkString("[", ", ", "]")
- throw ValidationException(
+ throw new ValidationException(
s"Field types of query result and registered TableSink " +
s"$sinkTableName do not match.\n" +
s"Query result schema: $srcSchema\n" +
@@ -855,7 +855,7 @@ abstract class TableEnvironment(val config: TableConfig) {
writeToSink(table, tableSink, conf)
case Some(_) =>
- throw TableException(s"The table registered as $sinkTableName is not a TableSink. " +
+ throw new TableException(s"The table registered as $sinkTableName is not a TableSink. " +
s"You can only emit query results to a registered TableSink.")
}
}
@@ -1119,7 +1119,7 @@ abstract class TableEnvironment(val config: TableConfig) {
// validate that at least the field types of physical and logical type match
// we do that here to make sure that plan translation was correct
if (schema.typeInfo != inputTypeInfo) {
- throw TableException(
+ throw new TableException(
s"The field types of physical and logical row types do not match. " +
s"Physical type is [${schema.typeInfo}], Logical type is [$inputTypeInfo]. " +
s"This is a bug and should not happen. Please file an issue.")
@@ -1350,7 +1350,8 @@ object TableEnvironment {
if ((clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) ||
!Modifier.isPublic(clazz.getModifiers) ||
clazz.getCanonicalName == null) {
- throw TableException(s"Class '$clazz' described in type information '$typeInfo' must be " +
+ throw new TableException(
+ s"Class '$clazz' described in type information '$typeInfo' must be " +
s"static and globally accessible.")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
index 81dad3e..0ca17a5 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/exceptions.scala
@@ -39,43 +39,6 @@ case class SqlParserException(
}
/**
- * General Exception for all errors during table handling.
- *
- * This exception indicates that an internal error occurred or that a feature is not supported
- * yet. Usually, this exception does not indicate a fault of the user.
- */
-case class TableException(
- msg: String,
- cause: Throwable)
- extends RuntimeException(msg, cause) {
-
- def this(msg: String) = this(msg, null)
-
-}
-
-object TableException {
- def apply(msg: String): TableException = new TableException(msg)
-}
-
-/**
- * Exception for all errors occurring during validation phase.
- *
- * This exception indicates that the user did something wrong.
- */
-case class ValidationException(
- msg: String,
- cause: Throwable)
- extends RuntimeException(msg, cause) {
-
- def this(msg: String) = this(msg, null)
-
-}
-
-object ValidationException {
- def apply(msg: String): ValidationException = new ValidationException(msg)
-}
-
-/**
* Exception for unwanted method calling on unresolved expression.
*/
case class UnresolvedException(msg: String) extends RuntimeException(msg)
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
index 2b9570c..dfb5c0b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
@@ -118,7 +118,7 @@ class Table(
val expandedFields = expandProjectList(fields, logicalPlan, tableEnv)
val (aggNames, propNames) = extractAggregationsAndProperties(expandedFields, tableEnv)
if (propNames.nonEmpty) {
- throw ValidationException("Window properties can only be used on windowed tables.")
+ throw new ValidationException("Window properties can only be used on windowed tables.")
}
if (aggNames.nonEmpty) {
@@ -822,14 +822,14 @@ class Table(
*/
def fetch(fetch: Int): Table = {
if (fetch < 0) {
- throw ValidationException("FETCH count must be equal or larger than 0.")
+ throw new ValidationException("FETCH count must be equal or larger than 0.")
}
this.logicalPlan match {
case Limit(o, -1, c) =>
// replace LIMIT without FETCH by LIMIT with FETCH
new Table(tableEnv, Limit(o, fetch, c).validate(tableEnv))
case Limit(_, _, _) =>
- throw ValidationException("FETCH is already defined.")
+ throw new ValidationException("FETCH is already defined.")
case _ =>
new Table(tableEnv, Limit(0, fetch, logicalPlan).validate(tableEnv))
}
@@ -934,7 +934,7 @@ class Table(
def insertInto(tableName: String, conf: QueryConfig): Unit = {
this.logicalPlan match {
case _: LogicalTableFunctionCall =>
- throw ValidationException("TableFunction can only be used in join and leftOuterJoin.")
+ throw new ValidationException("TableFunction can only be used in join and leftOuterJoin.")
case _ =>
tableEnv.insertInto(this, tableName, conf)
}
@@ -989,11 +989,11 @@ class Table(
def window(overWindows: OverWindow*): OverWindowedTable = {
if (tableEnv.isInstanceOf[BatchTableEnvironment]) {
- throw TableException("Over-windows for batch tables are currently not supported.")
+ throw new TableException("Over-windows for batch tables are currently not supported.")
}
if (overWindows.size != 1) {
- throw TableException("Over-Windows are currently only supported single window.")
+ throw new TableException("Over-Windows are currently only supported single window.")
}
new OverWindowedTable(this, overWindows.toArray)
@@ -1049,7 +1049,7 @@ class GroupedTable(
val expandedFields = expandProjectList(fields, table.logicalPlan, table.tableEnv)
val (aggNames, propNames) = extractAggregationsAndProperties(expandedFields, table.tableEnv)
if (propNames.nonEmpty) {
- throw ValidationException("Window properties can only be used on windowed tables.")
+ throw new ValidationException("Window properties can only be used on windowed tables.")
}
val projectsOnAgg = replaceAggregationsAndProperties(
@@ -1145,7 +1145,7 @@ class OverWindowedTable(
table.tableEnv)
if(fields.exists(_.isInstanceOf[WindowProperty])){
- throw ValidationException(
+ throw new ValidationException(
"Window start and end properties are not available for Over windows.")
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
index 2cd0c61..ca30053 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
@@ -133,7 +133,7 @@ class FlinkPlannerImpl(
// root = root.withRel(RelTimeIndicatorConverter.convert(root.rel, rexBuilder))
root
} catch {
- case e: RelConversionException => throw TableException(e.getMessage)
+ case e: RelConversionException => throw new TableException(e.getMessage)
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
index db7ffdb..c48967b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
@@ -172,7 +172,7 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
getTypeSystem.asInstanceOf[FlinkTypeSystem])
case ti@_ =>
- throw TableException(s"Unsupported type information: $ti")
+ throw new TableException(s"Unsupported type information: $ti")
}
canonize(relType)
@@ -307,7 +307,7 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
if (allTypes.exists(_.getSqlTypeName == SqlTypeName.ANY)) {
// one of the type was ANY.
// we cannot generate a common type if it differs from other types.
- throw TableException("Generic ANY types must have a common type information.")
+ throw new TableException("Generic ANY types must have a common type information.")
} else {
// cannot resolve a common type for different input types
None
@@ -341,10 +341,10 @@ object FlinkTypeFactory {
case TimeIntervalTypeInfo.INTERVAL_MILLIS => INTERVAL_DAY_SECOND
case CHAR_TYPE_INFO | CHAR_VALUE_TYPE_INFO =>
- throw TableException("Character type is not supported.")
+ throw new TableException("Character type is not supported.")
case _@t =>
- throw TableException(s"Type is not supported: $t")
+ throw new TableException(s"Type is not supported: $t")
}
/**
@@ -419,7 +419,8 @@ object FlinkTypeFactory {
case typeName if DAY_INTERVAL_TYPES.contains(typeName) => TimeIntervalTypeInfo.INTERVAL_MILLIS
case NULL =>
- throw TableException("Type NULL is not supported. Null values must have a supported type.")
+ throw new TableException(
+ "Type NULL is not supported. Null values must have a supported type.")
// symbol for special flags e.g. TRIM's BOTH, LEADING, TRAILING
// are represented as integer
@@ -454,6 +455,6 @@ object FlinkTypeFactory {
multisetRelDataType.typeInfo
case _@t =>
- throw TableException(s"Type is not supported: $t")
+ throw new TableException(s"Type is not supported: $t")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
index 2816f11..5d047ad 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
@@ -274,7 +274,7 @@ abstract class CodeGenerator(
generateExpression(rowtimeExpression.get)
case TimeIndicatorTypeInfo.ROWTIME_STREAM_MARKER |
TimeIndicatorTypeInfo.ROWTIME_BATCH_MARKER =>
- throw TableException("Rowtime extraction expression missing. Please report a bug.")
+ throw new TableException("Rowtime extraction expression missing. Please report a bug.")
case TimeIndicatorTypeInfo.PROCTIME_STREAM_MARKER =>
// attribute is proctime indicator.
// we use a null literal and generate a timestamp when we need it.
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/LiteralValueValidator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/LiteralValueValidator.scala
index 943b01a..f567131 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/LiteralValueValidator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/LiteralValueValidator.scala
@@ -62,7 +62,7 @@ class LiteralValueValidator(keyPrefix: String) extends HierarchyDescriptorValida
case Types.LONG => properties.validateLong(valueKey, isOptional = false)
case Types.SHORT => properties.validateShort(valueKey, isOptional = false)
case Types.STRING => properties.validateString(valueKey, isOptional = false)
- case _ => throw TableException(s"Unsupported type '$typeInfo'.")
+ case _ => throw new TableException(s"Unsupported type '$typeInfo'.")
}
}
// implicit type
@@ -107,7 +107,7 @@ object LiteralValueValidator {
case Types.LONG => properties.getLong(valueKey)
case Types.SHORT => properties.getShort(valueKey)
case Types.STRING => properties.getString(valueKey)
- case _ => throw TableException(s"Unsupported type '${typeInfo.getTypeClass}'.")
+ case _ => throw new TableException(s"Unsupported type '${typeInfo.getTypeClass}'.")
}
}
// implicit type
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/ExpressionUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/ExpressionUtils.scala
index e794d41..6080ba4 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/ExpressionUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/ExpressionUtils.scala
@@ -138,7 +138,7 @@ object ExpressionUtils {
if (array.length > 0 && array.head.isInstanceOf[Array[_]]) {
ArrayConstructor(array.map { na => convertArray(na.asInstanceOf[Array[_]]) })
} else {
- throw ValidationException("Unsupported array type.")
+ throw new ValidationException("Unsupported array type.")
}
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/call.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/call.scala
index 8454555..574cbb3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/call.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/call.scala
@@ -342,7 +342,7 @@ case class TableFunctionCall(
val fieldNames = if (aliases.isDefined) {
val aliasList = aliases.get
if (aliasList.length != originNames.length) {
- throw ValidationException(
+ throw new ValidationException(
s"List of column aliases must have same degree as table; " +
s"the returned table of function '$functionName' has ${originNames.length} " +
s"columns (${originNames.mkString(",")}), " +
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
index 3de0175..9721e62 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
@@ -190,7 +190,7 @@ case class RowtimeAttribute(expr: Expression) extends TimeAttribute(expr) {
// batch time window
Types.SQL_TIMESTAMP
case _ =>
- throw TableException("RowtimeAttribute has invalid type. Please report this bug.")
+ throw new TableException("RowtimeAttribute has invalid type. Please report this bug.")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
index 2c08001..9799e4d 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
@@ -53,15 +53,16 @@ object UserDefinedFunctionUtils {
*/
def checkForInstantiation(clazz: Class[_]): Unit = {
if (!InstantiationUtil.isPublic(clazz)) {
- throw ValidationException(s"Function class ${clazz.getCanonicalName} is not public.")
+ throw new ValidationException(s"Function class ${clazz.getCanonicalName} is not public.")
}
else if (!InstantiationUtil.isProperClass(clazz)) {
- throw ValidationException(s"Function class ${clazz.getCanonicalName} is no proper class," +
+ throw new ValidationException(
+ s"Function class ${clazz.getCanonicalName} is no proper class," +
" it is either abstract, an interface, or a primitive type.")
}
else if (InstantiationUtil.isNonStaticInnerClass(clazz)) {
- throw ValidationException(s"The class ${clazz.getCanonicalName} is an inner class, but" +
- " not statically accessible.")
+ throw new ValidationException(
+ s"The class ${clazz.getCanonicalName} is an inner class, but not statically accessible.")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
index 3475e19..53adf6b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
@@ -99,7 +99,7 @@ trait CommonCorrelate {
|}
|""".stripMargin
} else if (joinType != SemiJoinType.INNER) {
- throw TableException(s"Unsupported SemiJoinType: $joinType for correlate join.")
+ throw new TableException(s"Unsupported SemiJoinType: $joinType for correlate join.")
}
functionGenerator.generateFunction(
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
index ccfd904..56701f7 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
@@ -103,6 +103,6 @@ trait FlinkRelNode extends RelNode {
// 16 is an arbitrary estimate
(estimateDataTypeSize(t.getKeyType) + estimateDataTypeSize(t.getValueType)) * 16
case SqlTypeName.ANY => 128 // 128 is an arbitrary estimate
- case _ => throw TableException(s"Unsupported data type encountered: $t")
+ case _ => throw new TableException(s"Unsupported data type encountered: $t")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala
index 56bf8ea..7f32e6d 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetJoin.scala
@@ -128,7 +128,7 @@ class DataSetJoin(
val rightKeys = ArrayBuffer.empty[Int]
if (keyPairs.isEmpty) {
// if no equality keys => not supported
- throw TableException(
+ throw new TableException(
"Joins should have at least one equality condition.\n" +
s"\tLeft: ${left.toString},\n" +
s"\tRight: ${right.toString},\n" +
@@ -151,7 +151,7 @@ class DataSetJoin(
leftKeys.add(pair.source)
rightKeys.add(pair.target)
} else {
- throw TableException(
+ throw new TableException(
"Equality join predicate on incompatible types.\n" +
s"\tLeft: ${left.toString},\n" +
s"\tRight: ${right.toString},\n" +
@@ -255,7 +255,7 @@ class DataSetJoin(
config: TableConfig): DataSet[Row] = {
if (!config.getNullCheck) {
- throw TableException("Null check in TableConfig must be enabled for outer joins.")
+ throw new TableException("Null check in TableConfig must be enabled for outer joins.")
}
val joinOpName = getJoinOpName
@@ -311,7 +311,7 @@ class DataSetJoin(
config: TableConfig): DataSet[Row] = {
if (!config.getNullCheck) {
- throw TableException("Null check in TableConfig must be enabled for outer joins.")
+ throw new TableException("Null check in TableConfig must be enabled for outer joins.")
}
val joinOpName = getJoinOpName
@@ -367,7 +367,7 @@ class DataSetJoin(
config: TableConfig): DataSet[Row] = {
if (!config.getNullCheck) {
- throw TableException("Null check in TableConfig must be enabled for outer joins.")
+ throw new TableException("Null check in TableConfig must be enabled for outer joins.")
}
val joinOpName = getJoinOpName
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSort.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSort.scala
index 1430204..fae52fa 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSort.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSort.scala
@@ -84,7 +84,7 @@ class DataSetSort(
queryConfig: BatchQueryConfig): DataSet[Row] = {
if (fieldCollations.isEmpty) {
- throw TableException("Limiting the result without sorting is not allowed " +
+ throw new TableException("Limiting the result without sorting is not allowed " +
"as it could lead to arbitrary results.")
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
index 0a014b6..78a7273 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
@@ -147,7 +147,7 @@ class DataStreamGroupWindowAggregate(
val timeAttribute = window.timeAttribute.asInstanceOf[ResolvedFieldReference].name
val timeIdx = inputSchema.fieldNames.indexOf(timeAttribute)
if (timeIdx < 0) {
- throw TableException("Time attribute could not be found. This is a bug.")
+ throw new TableException("Time attribute could not be found. This is a bug.")
}
inputDS
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
index 3a38bc5..80c3123 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
@@ -134,7 +134,7 @@ class DataStreamJoin(
val rightKeyType = rightFields.get(pair.target).getType.getSqlTypeName
// check if keys are compatible
if (leftKeyType != rightKeyType) {
- throw TableException(
+ throw new TableException(
"Equality join predicate on incompatible types.\n" +
s"\tLeft: $left,\n" +
s"\tRight: $right,\n" +
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamMatch.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamMatch.scala
index 2bf6e32..19b298c 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamMatch.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamMatch.scala
@@ -292,16 +292,16 @@ private[flink] class PatternVisitor(
applyQuantifier(pattern, startNum, endNum, isGreedy)
case PATTERN_ALTER =>
- throw TableException(s"Expression not supported: $call. Currently, CEP doesn't support " +
- s"branching patterns.")
+ throw new TableException(
+ s"Expression not supported: $call. Currently, CEP doesn't support branching patterns.")
case PATTERN_PERMUTE =>
- throw TableException(s"Expression not supported: $call. Currently, CEP doesn't support " +
- s"PERMUTE patterns.")
+ throw new TableException(
+ s"Expression not supported: $call. Currently, CEP doesn't support PERMUTE patterns.")
case PATTERN_EXCLUDE =>
- throw TableException(s"Expression not supported: $call. Currently, CEP doesn't support " +
- s"'{-' '-}' patterns.")
+ throw new TableException(
+ s"Expression not supported: $call. Currently, CEP doesn't support '{-' '-}' patterns.")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/LogicalWindowAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/LogicalWindowAggregateRule.scala
index 11701fc..8789ba6 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/LogicalWindowAggregateRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/LogicalWindowAggregateRule.scala
@@ -114,19 +114,19 @@ abstract class LogicalWindowAggregateRule(ruleName: String)
if (call.getOperands.size() == 2) {
true
} else {
- throw TableException("TUMBLE window with alignment is not supported yet.")
+ throw new TableException("TUMBLE window with alignment is not supported yet.")
}
case BasicOperatorTable.HOP =>
if (call.getOperands.size() == 3) {
true
} else {
- throw TableException("HOP window with alignment is not supported yet.")
+ throw new TableException("HOP window with alignment is not supported yet.")
}
case BasicOperatorTable.SESSION =>
if (call.getOperands.size() == 2) {
true
} else {
- throw TableException("SESSION window with alignment is not supported yet.")
+ throw new TableException("SESSION window with alignment is not supported yet.")
}
case _ => false
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowPropertiesRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowPropertiesRule.scala
index 74f7b1b..b9e34ff 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowPropertiesRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowPropertiesRule.scala
@@ -145,7 +145,7 @@ abstract class WindowPropertiesBaseRule(rulePredicate: RelOptRuleOperand, ruleNa
// replace expression by access to window rowtime
builder.field(propertyName(window, "rowtime"))
case 'streamProctime =>
- throw ValidationException("A proctime window cannot provide a rowtime attribute.")
+ throw new ValidationException("A proctime window cannot provide a rowtime attribute.")
case _ =>
throw new TableException("Unknown window type encountered. Please report this bug.")
}
@@ -156,7 +156,8 @@ abstract class WindowPropertiesBaseRule(rulePredicate: RelOptRuleOperand, ruleNa
// replace expression by access to window proctime
builder.field(propertyName(window, "proctime"))
case 'batchRowtime =>
- throw ValidationException("PROCTIME window property is not supported in batch queries.")
+ throw new ValidationException(
+ "PROCTIME window property is not supported in batch queries.")
case _ =>
throw new TableException("Unknown window type encountered. Please report this bug.")
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala
index f215329..72d71aa 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala
@@ -40,13 +40,13 @@ class DataSetWindowAggregateRule
// check if we have distinct aggregates
val distinctAggs = agg.getAggCallList.exists(_.isDistinct)
if (distinctAggs) {
- throw TableException("DISTINCT aggregates are currently not supported.")
+ throw new TableException("DISTINCT aggregates are currently not supported.")
}
// check if we have grouping sets
val groupSets = agg.getGroupSets.size() != 1 || agg.getGroupSets.get(0) != agg.getGroupSet
if (groupSets || agg.indicator) {
- throw TableException("GROUPING SETS are currently not supported.")
+ throw new TableException("GROUPING SETS are currently not supported.")
}
!distinctAggs && !groupSets && !agg.indicator
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupAggregateRule.scala
index e99a86e..c80a145 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupAggregateRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupAggregateRule.scala
@@ -46,7 +46,7 @@ class DataStreamGroupAggregateRule
// check if we have grouping sets
val groupSets = agg.getGroupSets.size() != 1 || agg.getGroupSets.get(0) != agg.getGroupSet
if (groupSets || agg.indicator) {
- throw TableException("GROUPING SETS are currently not supported.")
+ throw new TableException("GROUPING SETS are currently not supported.")
}
!groupSets && !agg.indicator
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupWindowAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupWindowAggregateRule.scala
index b77b7bf..d48beaf 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupWindowAggregateRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupWindowAggregateRule.scala
@@ -42,7 +42,7 @@ class DataStreamGroupWindowAggregateRule
// check if we have grouping sets
val groupSets = agg.getGroupSets.size() != 1 || agg.getGroupSets.get(0) != agg.getGroupSet
if (groupSets || agg.indicator) {
- throw TableException("GROUPING SETS are currently not supported.")
+ throw new TableException("GROUPING SETS are currently not supported.")
}
!groupSets && !agg.indicator
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamLogicalWindowAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamLogicalWindowAggregateRule.scala
index 254f36b..62eecbb 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamLogicalWindowAggregateRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamLogicalWindowAggregateRule.scala
@@ -46,7 +46,7 @@ class DataStreamLogicalWindowAggregateRule
timeAttribute
case _ =>
- throw TableException(
+ throw new TableException(
s"Time attribute expected but ${timeAttribute.getType} encountered.")
}
}
@@ -69,7 +69,7 @@ class DataStreamLogicalWindowAggregateRule
def getOperandAsLong(call: RexCall, idx: Int): Long =
call.getOperands.get(idx) match {
case v: RexLiteral => v.getValue.asInstanceOf[JBigDecimal].longValue()
- case _ => throw TableException("Only constant window descriptors are supported.")
+ case _ => throw new TableException("Only constant window descriptors are supported.")
}
def getOperandAsTimeIndicator(call: RexCall, idx: Int): ResolvedFieldReference =
@@ -79,7 +79,7 @@ class DataStreamLogicalWindowAggregateRule
rowType.getFieldList.get(v.getIndex).getName,
FlinkTypeFactory.toTypeInfo(v.getType))
case _ =>
- throw ValidationException("Window can only be defined over a time attribute column.")
+ throw new ValidationException("Window can only be defined over a time attribute column.")
}
windowExpr.getOperator match {
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
index f985a80..09886e0 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
@@ -96,7 +96,7 @@ class LogicalUnnestRule(
ExplodeFunctionUtil.explodeTableFuncFromType(arrayType.typeInfo))
case mt: MultisetRelDataType =>
(mt.getComponentType, ExplodeFunctionUtil.explodeTableFuncFromType(mt.typeInfo))
- case _ => throw TableException(s"Unsupported UNNEST on type: ${dataType.toString}")
+ case _ => throw new TableException(s"Unsupported UNNEST on type: ${dataType.toString}")
}
// create sql function
@@ -121,7 +121,7 @@ class LogicalUnnestRule(
StructKind.FULLY_QUALIFIED,
ImmutableList.of(new RelDataTypeFieldImpl("f0", 0, componentType)))
case _: RelRecordType => componentType
- case _ => throw TableException(
+ case _ => throw new TableException(
s"Unsupported component type in UNNEST: ${componentType.toString}")
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
index a3c9c1e..1e2df6e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
@@ -1146,15 +1146,18 @@ object AggregateUtil {
case NamedWindowProperty(_, prop) =>
prop match {
case WindowStart(_) if s.isDefined =>
- throw TableException("Duplicate window start property encountered. This is a bug.")
+ throw new TableException(
+ "Duplicate window start property encountered. This is a bug.")
case WindowStart(_) =>
(Some(i), e, rt, i - 1)
case WindowEnd(_) if e.isDefined =>
- throw TableException("Duplicate window end property encountered. This is a bug.")
+ throw new TableException(
+ "Duplicate window end property encountered. This is a bug.")
case WindowEnd(_) =>
(s, Some(i), rt, i - 1)
case RowtimeAttribute(_) if rt.isDefined =>
- throw TableException("Duplicate window rowtime property encountered. This is a bug.")
+ throw new TableException(
+ "Duplicate window rowtime property encountered. This is a bug.")
case RowtimeAttribute(_) =>
(s, e, Some(i), i - 1)
case ProctimeAttribute(_) =>
@@ -1550,10 +1553,10 @@ object AggregateUtil {
if (relDataType.length == 1) {
relDataType.head.getIndex
} else {
- throw TableException(
+ throw new TableException(
s"Encountered more than one time attribute with the same name: '$relDataType'")
}
- case e => throw TableException(
+ case e => throw new TableException(
"The time attribute of window in batch environment should be " +
s"ResolvedFieldReference, but is $e")
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSource.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSource.scala
index e0022c5..f7215ee 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSource.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/CsvTableSource.scala
@@ -115,7 +115,7 @@ class CsvTableSource private (
}
if (fieldNames.length != fieldTypes.length) {
- throw TableException("Number of field names and field types must be equal.")
+ throw new TableException("Number of field names and field types must be equal.")
}
private val selectedFieldTypes = selectedFields.map(fieldTypes(_))
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
index 5cb7e90..ef1aa16 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/TableSourceUtil.scala
@@ -89,14 +89,14 @@ object TableSourceUtil {
val (physicalName, _, tpe) = resolveInputField(name, tableSource)
// validate that mapped fields are are same type
if (tpe != t) {
- throw ValidationException(s"Type $t of table field '$name' does not " +
+ throw new ValidationException(s"Type $t of table field '$name' does not " +
s"match with type $tpe of the field '$physicalName' of the TableSource return type.")
}
mappedFieldCnt += 1
}
// ensure that only one field is mapped to an atomic type
if (!tableSource.getReturnType.isInstanceOf[CompositeType[_]] && mappedFieldCnt > 1) {
- throw ValidationException(
+ throw new ValidationException(
s"More than one table field matched to atomic input type ${tableSource.getReturnType}.")
}
@@ -105,7 +105,8 @@ object TableSourceUtil {
case r: DefinedRowtimeAttributes =>
val descriptors = r.getRowtimeAttributeDescriptors
if (descriptors.size() > 1) {
- throw ValidationException("Currently, only a single rowtime attribute is supported. " +
+ throw new ValidationException(
+ "Currently, only a single rowtime attribute is supported. " +
s"Please remove all but one RowtimeAttributeDescriptor.")
} else if (descriptors.size() == 1) {
val descriptor = descriptors.get(0)
@@ -113,12 +114,12 @@ object TableSourceUtil {
val rowtimeIdx = schema.getColumnNames.indexOf(rowtimeAttribute)
// ensure that field exists
if (rowtimeIdx < 0) {
- throw ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
+ throw new ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
s"'$rowtimeAttribute' but field '$rowtimeAttribute' does not exist in table.")
}
// ensure that field is of type TIMESTAMP
if (schema.getTypes(rowtimeIdx) != Types.SQL_TIMESTAMP) {
- throw ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
+ throw new ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
s"'$rowtimeAttribute' but field '$rowtimeAttribute' is not of type TIMESTAMP.")
}
// look up extractor input fields in return type
@@ -137,12 +138,12 @@ object TableSourceUtil {
val proctimeIdx = schema.getColumnNames.indexOf(proctimeAttribute)
// ensure that field exists
if (proctimeIdx < 0) {
- throw ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
+ throw new ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
s"'$proctimeAttribute' but field '$proctimeAttribute' does not exist in table.")
}
// ensure that field is of type TIMESTAMP
if (schema.getTypes(proctimeIdx) != Types.SQL_TIMESTAMP) {
- throw ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
+ throw new ValidationException(s"Found a RowtimeAttributeDescriptor for field " +
s"'$proctimeAttribute' but field '$proctimeAttribute' is not of type TIMESTAMP.")
}
case _ => // nothing to validate
@@ -224,7 +225,7 @@ object TableSourceUtil {
val (physicalName, idx, tpe) = resolveInputField(name, tableSource)
// validate that mapped fields are are same type
if (tpe != t) {
- throw ValidationException(s"Type $t of table field '$name' does not " +
+ throw new ValidationException(s"Type $t of table field '$name' does not " +
s"match with type $tpe of the field '$physicalName' of the TableSource return type.")
}
idx
@@ -232,7 +233,7 @@ object TableSourceUtil {
// ensure that only one field is mapped to an atomic type
if (!inputType.isInstanceOf[CompositeType[_]] && mapping.count(_ >= 0) > 1) {
- throw ValidationException(
+ throw new ValidationException(
s"More than one table field matched to atomic input type $inputType.")
}
@@ -302,7 +303,7 @@ object TableSourceUtil {
if (descriptors.size() == 0) {
None
} else if (descriptors.size > 1) {
- throw ValidationException("Table with has more than a single rowtime attribute..")
+ throw new ValidationException("Table with has more than a single rowtime attribute.")
} else {
// exactly one rowtime attribute descriptor
if (selectedFields.isEmpty) {
@@ -418,7 +419,7 @@ object TableSourceUtil {
val accessedFields = if (rowtimeAttributeDescriptor.isDefined) {
rowtimeAttributeDescriptor.get.getTimestampExtractor.getArgumentFields
} else {
- throw TableException("Computed field mapping includes a rowtime marker but the " +
+ throw new TableException("Computed field mapping includes a rowtime marker but the " +
"TableSource does not provide a RowtimeAttributeDescriptor. " +
"This is a bug and should be reported.")
}
@@ -471,7 +472,7 @@ object TableSourceUtil {
// get and check field index
val idx = c.getFieldIndex(fieldName)
if (idx < 0) {
- throw ValidationException(failMsg)
+ throw new ValidationException(failMsg)
}
// return field name, index, and field type
(fieldName, idx, c.getTypeAt(idx))
@@ -487,7 +488,7 @@ object TableSourceUtil {
// resolve field name in field mapping
val resolvedFieldName = d.getFieldMapping.get(fieldName)
if (resolvedFieldName == null) {
- throw ValidationException(
+ throw new ValidationException(
s"Field '$fieldName' could not be resolved by the field mapping.")
}
// look up resolved field in return type
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala
index 979f869..2468b1a 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala
@@ -42,7 +42,7 @@ final class ExistingField(val field: String) extends TimestampExtractor {
case Types.SQL_TIMESTAMP => // OK
case Types.STRING => // OK
case _: TypeInformation[_] =>
- throw ValidationException(
+ throw new ValidationException(
s"Field '$field' must be of type Long or Timestamp or String but is of type $fieldType.")
}
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
index ff7fbf5..efcd006 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
@@ -64,7 +64,7 @@ class FunctionCatalog {
*/
def lookupFunction(name: String, children: Seq[Expression]): Expression = {
val funcClass = functionBuilders
- .getOrElse(name.toLowerCase, throw ValidationException(s"Undefined function: $name"))
+ .getOrElse(name.toLowerCase, throw new ValidationException(s"Undefined function: $name"))
// Instantiate a function using the provided `children`
funcClass match {
@@ -73,7 +73,7 @@ class FunctionCatalog {
case sf if classOf[ScalarFunction].isAssignableFrom(sf) =>
val scalarSqlFunction = sqlFunctions
.find(f => f.getName.equalsIgnoreCase(name) && f.isInstanceOf[ScalarSqlFunction])
- .getOrElse(throw ValidationException(s"Undefined scalar function: $name"))
+ .getOrElse(throw new ValidationException(s"Undefined scalar function: $name"))
.asInstanceOf[ScalarSqlFunction]
ScalarFunctionCall(scalarSqlFunction.getScalarFunction, children)
@@ -81,7 +81,7 @@ class FunctionCatalog {
case tf if classOf[TableFunction[_]].isAssignableFrom(tf) =>
val tableSqlFunction = sqlFunctions
.find(f => f.getName.equalsIgnoreCase(name) && f.isInstanceOf[TableSqlFunction])
- .getOrElse(throw ValidationException(s"Undefined table function: $name"))
+ .getOrElse(throw new ValidationException(s"Undefined table function: $name"))
.asInstanceOf[TableSqlFunction]
val typeInfo = tableSqlFunction.getRowTypeInfo
val function = tableSqlFunction.getTableFunction
@@ -91,7 +91,7 @@ class FunctionCatalog {
case af if classOf[AggregateFunction[_, _]].isAssignableFrom(af) =>
val aggregateFunction = sqlFunctions
.find(f => f.getName.equalsIgnoreCase(name) && f.isInstanceOf[AggSqlFunction])
- .getOrElse(throw ValidationException(s"Undefined table function: $name"))
+ .getOrElse(throw new ValidationException(s"Undefined table function: $name"))
.asInstanceOf[AggSqlFunction]
val function = aggregateFunction.getFunction
val returnType = aggregateFunction.returnType
@@ -121,16 +121,16 @@ class FunctionCatalog {
case Success(ctor) =>
Try(ctor.newInstance(children: _*).asInstanceOf[Expression]) match {
case Success(expr) => expr
- case Failure(exception) => throw ValidationException(exception.getMessage)
+ case Failure(exception) => throw new ValidationException(exception.getMessage)
}
case Failure(_) =>
- throw ValidationException(
+ throw new ValidationException(
s"Invalid number of arguments for function $funcClass")
}
}
}
case _ =>
- throw ValidationException("Unsupported function.")
+ throw new ValidationException("Unsupported function.")
}
}
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/RowtimeTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/RowtimeTest.scala
index 199d416..02ff5ce 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/RowtimeTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/RowtimeTest.scala
@@ -124,7 +124,7 @@ object RowtimeTest {
argumentFieldTypes(0) match {
case Types.SQL_TIMESTAMP =>
case _ =>
- throw ValidationException(
+ throw new ValidationException(
s"Field 'ts' must be of type Timestamp but is of type ${argumentFieldTypes(0)}.")
}
}