This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 1904b215e36 [FLINK-35155] Introduce TableRuntimeException (#24679)
1904b215e36 is described below
commit 1904b215e36e4fd48e48ece7ffdf2f1470653130
Author: Dawid Wysakowicz <[email protected]>
AuthorDate: Mon May 6 11:00:53 2024 +0200
[FLINK-35155] Introduce TableRuntimeException (#24679)
---
.../TableRuntimeException.java} | 34 ++++-----
.../flink/table/data/utils/CastExecutor.java | 4 +-
.../casting/AbstractCodeGeneratorCastRule.java | 6 +-
.../table/planner/codegen/ExprCodeGenerator.scala | 2 +-
.../BuiltInAggregateFunctionTestBase.java | 79 ++++++++++++++++++-
.../planner/functions/BuiltInFunctionTestBase.java | 15 ++++
.../planner/functions/JsonFunctionsITCase.java | 7 ++
.../planner/functions/MiscAggFunctionITCase.java | 54 +++++++++++++
.../planner/functions/casting/CastRulesTest.java | 89 +++++++++++-----------
.../table/runtime/functions/SqlJsonUtils.java | 42 +++++-----
10 files changed, 239 insertions(+), 93 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/CastExecutor.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableRuntimeException.java
similarity index 54%
copy from
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/CastExecutor.java
copy to
flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableRuntimeException.java
index e701a198533..fffbedbd629 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/CastExecutor.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableRuntimeException.java
@@ -16,28 +16,24 @@
* limitations under the License.
*/
-package org.apache.flink.table.data.utils;
+package org.apache.flink.table.api;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.TableException;
-
-import javax.annotation.Nullable;
+import org.apache.flink.annotation.PublicEvolving;
/**
- * Interface to model a function that performs the casting of a value from one
type to another.
+ * Exception for errors occurring in the runtime.
*
- * @param <IN> Input internal type
- * @param <OUT> Output internal type
+ * <p>This exception indicates the exception was thrown intentionally, e.g.
during evaluation of
+ * {@code SINGLE_VALUE} function. Most likely a user error.
*/
-@Internal
-@FunctionalInterface
-public interface CastExecutor<IN, OUT> {
- /**
- * Cast the input value. The output is null only and only if the input is
null. The method
- * throws an exception if something goes wrong when casting.
- *
- * @param value Input value.
- */
- @Nullable
- OUT cast(@Nullable IN value) throws TableException;
+@PublicEvolving
+public class TableRuntimeException extends RuntimeException {
+
+ public TableRuntimeException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public TableRuntimeException(String message) {
+ super(message);
+ }
}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/CastExecutor.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/CastExecutor.java
index e701a198533..032b68dda67 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/CastExecutor.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/CastExecutor.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.data.utils;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableRuntimeException;
import javax.annotation.Nullable;
@@ -39,5 +39,5 @@ public interface CastExecutor<IN, OUT> {
* @param value Input value.
*/
@Nullable
- OUT cast(@Nullable IN value) throws TableException;
+ OUT cast(@Nullable IN value) throws TableRuntimeException;
}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
index 7eabc4d4f12..71cdaa33563 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.planner.functions.casting;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.data.utils.CastExecutor;
import org.apache.flink.table.planner.codegen.CodeGenUtils;
import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
@@ -112,7 +112,7 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT>
extends AbstractCastRule<I
// the cast method
final String functionSignature =
"@Override public Object cast(Object _myInputObj) throws "
- + className(TableException.class);
+ + className(TableRuntimeException.class);
// Write the function body
final CastRuleUtils.CodeWriter bodyWriter = new
CastRuleUtils.CodeWriter();
@@ -127,7 +127,7 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT>
extends AbstractCastRule<I
(exceptionTerm, catchWriter) ->
catchWriter.throwStmt(
constructorCall(
- TableException.class,
+ TableRuntimeException.class,
strLiteral(
"Error when casting "
+ inputLogicalType
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
index 07d45f396fb..82e9951864b 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
@@ -773,7 +773,7 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext,
nullableInput: Boolean)
|${operands.map(_.code).mkString("\n")}
|${nullValue.code}
|org.apache.flink.util.ExceptionUtils.rethrow(
- | new RuntimeException(${operands.head.resultTerm}.toString()));
+ | new
org.apache.flink.table.api.TableRuntimeException(${operands.head.resultTerm}.toString()));
|""".stripMargin
GeneratedExpression(nullValue.resultTerm, nullValue.nullTerm, code,
resultType)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
index dd05dad760a..bbb2d24fcb2 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
@@ -42,7 +42,7 @@ import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.CloseableIterator;
-import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.CollectionUtil;
import org.jetbrains.annotations.NotNull;
import org.junit.jupiter.api.TestInstance;
@@ -69,6 +69,7 @@ import static
org.apache.flink.runtime.state.StateBackendLoader.ROCKSDB_STATE_BA
import static org.apache.flink.table.test.TableAssertions.assertThat;
import static org.apache.flink.table.types.DataType.getFieldDataTypes;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test base for testing aggregate {@link BuiltInFunctionDefinition built-in
functions}. */
@Execution(ExecutionMode.CONCURRENT)
@@ -176,7 +177,7 @@ abstract class BuiltInAggregateFunctionTestBase {
/** Test specification. */
protected static class TestSpec {
- private final BuiltInFunctionDefinition definition;
+ private final @Nullable BuiltInFunctionDefinition definition;
private final List<TestItem> testItems = new ArrayList<>();
private @Nullable String description;
@@ -185,13 +186,17 @@ abstract class BuiltInAggregateFunctionTestBase {
private List<Row> sourceRows;
private TestSpec(BuiltInFunctionDefinition definition) {
- this.definition = Preconditions.checkNotNull(definition);
+ this.definition = definition;
}
static TestSpec forFunction(BuiltInFunctionDefinition definition) {
return new TestSpec(definition);
}
+ static TestSpec forExpression(String expr) {
+ return new TestSpec(null).withDescription(expr);
+ }
+
TestSpec withDescription(String description) {
this.description = description;
return this;
@@ -209,6 +214,17 @@ abstract class BuiltInAggregateFunctionTestBase {
return this;
}
+ TestSpec testSqlRuntimeError(
+ Function<Table, String> sqlSpec,
+ DataType expectedRowType,
+ Class<? extends Throwable> exceptionClass,
+ String exceptionMessage) {
+ this.testItems.add(
+ new SqlRuntimeErrorItem(
+ sqlSpec, expectedRowType, exceptionClass,
exceptionMessage));
+ return this;
+ }
+
TestSpec testApiResult(
List<Expression> selectExpr,
List<Expression> groupByExpr,
@@ -297,7 +313,9 @@ abstract class BuiltInAggregateFunctionTestBase {
@Override
public String toString() {
final StringBuilder bob = new StringBuilder();
- bob.append(definition.getName());
+ if (definition != null) {
+ bob.append(definition.getName());
+ }
if (description != null) {
bob.append(" (");
bob.append(description);
@@ -344,6 +362,41 @@ abstract class BuiltInAggregateFunctionTestBase {
protected abstract TableResult getResult(TableEnvironment tEnv, Table
sourceTable);
}
+ private abstract static class RuntimeErrorItem implements TestItem {
+ private final @Nullable DataType expectedRowType;
+ private final Class<? extends Throwable> exceptionClass;
+ private final String exceptionMessage;
+
+ public RuntimeErrorItem(
+ @Nullable DataType expectedRowType,
+ Class<? extends Throwable> exceptionClass,
+ String exceptionMessage) {
+ this.expectedRowType = expectedRowType;
+ this.exceptionClass = exceptionClass;
+ this.exceptionMessage = exceptionMessage;
+ }
+
+ @Override
+ public void execute(TableEnvironment tEnv, Table sourceTable) {
+ final TableResult tableResult = getResult(tEnv, sourceTable);
+
+ if (expectedRowType != null) {
+ final DataType actualRowType =
+ tableResult.getResolvedSchema().toSourceRowDataType();
+
+ assertThat(actualRowType)
+ .getChildren()
+
.containsExactlyElementsOf(getFieldDataTypes(expectedRowType));
+ }
+
+ assertThatThrownBy(() ->
CollectionUtil.iteratorToList(tableResult.collect()))
+ .hasRootCauseInstanceOf(exceptionClass)
+ .hasRootCauseMessage(exceptionMessage);
+ }
+
+ protected abstract TableResult getResult(TableEnvironment tEnv, Table
sourceTable);
+ }
+
private static class SqlTestItem extends SuccessItem {
private final Function<Table, String> spec;
@@ -361,6 +414,24 @@ abstract class BuiltInAggregateFunctionTestBase {
}
}
+ private static class SqlRuntimeErrorItem extends RuntimeErrorItem {
+ private final Function<Table, String> spec;
+
+ public SqlRuntimeErrorItem(
+ Function<Table, String> spec,
+ @Nullable DataType expectedRowType,
+ Class<? extends Throwable> exceptionClass,
+ String exceptionMessage) {
+ super(expectedRowType, exceptionClass, exceptionMessage);
+ this.spec = spec;
+ }
+
+ @Override
+ protected TableResult getResult(TableEnvironment tEnv, Table
sourceTable) {
+ return tEnv.sqlQuery(spec.apply(sourceTable)).execute();
+ }
+ }
+
private static class TableApiTestItem extends SuccessItem {
private final List<Expression> selectExpr;
private final List<Expression> groupByExpr;
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInFunctionTestBase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInFunctionTestBase.java
index 96e49e96f47..915aba42856 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInFunctionTestBase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInFunctionTestBase.java
@@ -216,6 +216,15 @@ abstract class BuiltInFunctionTestBase {
return this;
}
+ TestSetSpec testTableApiRuntimeError(
+ Expression expression,
+ Class<? extends Throwable> exceptionError,
+ String errorMessage) {
+ testItems.add(
+ new TableApiErrorTestItem(expression, exceptionError,
errorMessage, false));
+ return this;
+ }
+
TestSetSpec testSqlResult(String expression, Object result,
AbstractDataType<?> dataType) {
return testSqlResult(expression, singletonList(result),
singletonList(dataType));
}
@@ -248,6 +257,12 @@ abstract class BuiltInFunctionTestBase {
return this;
}
+ TestSetSpec testSqlRuntimeError(
+ String expression, Class<? extends Throwable> exceptionError,
String errorMessage) {
+ testItems.add(new SqlErrorTestItem(expression, exceptionError,
errorMessage, false));
+ return this;
+ }
+
TestSetSpec testResult(
Expression expression,
String sqlExpression,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
index 1f5fa406a7f..4d34488cdc1 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.api.JsonExistsOnError;
import org.apache.flink.table.api.JsonOnNull;
import org.apache.flink.table.api.JsonType;
import org.apache.flink.table.api.JsonValueOnEmptyOrError;
+import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
@@ -162,9 +163,11 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
BOOLEAN())
.testSqlRuntimeError(
"JSON_EXISTS(f0, 'strict $.invalid' ERROR ON ERROR)",
+ TableRuntimeException.class,
"No results for path: $['invalid']")
.testTableApiRuntimeError(
$("f0").jsonExists("strict $.invalid",
JsonExistsOnError.ERROR),
+ TableRuntimeException.class,
"No results for path: $['invalid']");
}
@@ -400,9 +403,11 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
STRING())
.testSqlRuntimeError(
"JSON_QUERY(f0, 'lax $.err4' ERROR ON EMPTY)",
+ TableRuntimeException.class,
"Empty result of JSON_QUERY function is not
allowed")
.testTableApiRuntimeError(
$("f0").jsonQuery("lax $.err5", WITHOUT_ARRAY,
ERROR, NULL),
+ TableRuntimeException.class,
"Empty result of JSON_QUERY function is not
allowed")
// Error Behavior
@@ -426,9 +431,11 @@ class JsonFunctionsITCase extends BuiltInFunctionTestBase {
STRING())
.testSqlRuntimeError(
"JSON_QUERY(f0, 'strict $.err9' ERROR ON
ERROR)",
+ TableRuntimeException.class,
"No results for path")
.testTableApiRuntimeError(
$("f0").jsonQuery("strict $.err10",
WITHOUT_ARRAY, NULL, ERROR),
+ TableRuntimeException.class,
"No results for path"));
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MiscAggFunctionITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MiscAggFunctionITCase.java
new file mode 100644
index 00000000000..50a7ca55e11
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MiscAggFunctionITCase.java
@@ -0,0 +1,54 @@
+/*
+ * 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.planner.functions;
+
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.types.Row;
+
+import java.util.Arrays;
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.types.RowKind.INSERT;
+
+/** Tests for built-in ARRAY_AGG aggregation functions. */
+class MiscAggFunctionITCase extends BuiltInAggregateFunctionTestBase {
+
+ @Override
+ Stream<TestSpec> getTestCaseSpecs() {
+ return Stream.of(
+ TestSpec.forExpression("SINGLE_VALUE")
+ .withSource(
+ ROW(STRING(), INT()),
+ Arrays.asList(
+ Row.ofKind(INSERT, "A", 1),
+ Row.ofKind(INSERT, "A", 2),
+ Row.ofKind(INSERT, "B", 2)))
+ .testSqlRuntimeError(
+ source ->
+ "SELECT f0, SINGLE_VALUE(f1) FROM "
+ + source
+ + " GROUP BY f0",
+ ROW(STRING(), INT()),
+ TableRuntimeException.class,
+ "SingleValueAggFunction received more than one
element."));
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java
index 459c94a46a4..c6d634577ca 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java
@@ -22,7 +22,7 @@ import
org.apache.flink.api.common.typeutils.base.LocalDateSerializer;
import org.apache.flink.api.common.typeutils.base.LocalDateTimeSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.GenericArrayData;
@@ -170,13 +170,13 @@ class CastRulesTest {
return Stream.of(
CastTestSpecBuilder.testCastTo(TINYINT())
.fromCase(TINYINT(), null, null)
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("Apache"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("Apache"),
TableRuntimeException.class)
.fromCase(STRING(), fromString("1.234"), (byte) 1)
.fromCase(STRING(), fromString("123"), (byte) 123)
.fromCase(STRING(), fromString(" 123 "), (byte) 123)
- .fail(STRING(), fromString("-130"),
TableException.class)
+ .fail(STRING(), fromString("-130"),
TableRuntimeException.class)
.fromCase(
DECIMAL(4, 3),
fromBigDecimal(new BigDecimal("9.87"), 4, 3),
@@ -205,13 +205,13 @@ class CastRulesTest {
.fromCase(BOOLEAN(), false, (byte) 0),
CastTestSpecBuilder.testCastTo(SMALLINT())
.fromCase(SMALLINT(), null, null)
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("Apache"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("Apache"),
TableRuntimeException.class)
.fromCase(STRING(), fromString("1.234"), (short) 1)
.fromCase(STRING(), fromString("123"), (short) 123)
.fromCase(STRING(), fromString(" 123 "), (short) 123)
- .fail(STRING(), fromString("-32769"),
TableException.class)
+ .fail(STRING(), fromString("-32769"),
TableRuntimeException.class)
.fromCase(
DECIMAL(4, 3),
fromBigDecimal(new BigDecimal("9.87"), 4, 3),
@@ -250,13 +250,13 @@ class CastRulesTest {
.fromCase(BOOLEAN(), true, (short) 1)
.fromCase(BOOLEAN(), false, (short) 0),
CastTestSpecBuilder.testCastTo(INT())
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("Apache"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("Apache"),
TableRuntimeException.class)
.fromCase(STRING(), fromString("1.234"), 1)
.fromCase(STRING(), fromString("123"), 123)
.fromCase(STRING(), fromString(" 123 "), 123)
- .fail(STRING(), fromString("-3276913443134"),
TableException.class)
+ .fail(STRING(), fromString("-3276913443134"),
TableRuntimeException.class)
.fromCase(DECIMAL(4, 3), fromBigDecimal(new
BigDecimal("9.87"), 4, 3), 9)
// https://issues.apache.org/jira/browse/FLINK-24420 -
Check out of range
// instead of overflow
@@ -297,9 +297,9 @@ class CastRulesTest {
.fromCase(BOOLEAN(), false, 0),
CastTestSpecBuilder.testCastTo(BIGINT())
.fromCase(BIGINT(), null, null)
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("Apache"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("Apache"),
TableRuntimeException.class)
.fromCase(STRING(), fromString("1.234"), 1L)
.fromCase(STRING(), fromString("123"), 123L)
.fromCase(STRING(), fromString(" 123 "), 123L)
@@ -339,9 +339,9 @@ class CastRulesTest {
.fromCase(BOOLEAN(), false, 0L),
CastTestSpecBuilder.testCastTo(FLOAT())
.fromCase(FLOAT(), null, null)
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("Apache"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("Apache"),
TableRuntimeException.class)
.fromCase(STRING(), fromString("1.234"), 1.234f)
.fromCase(STRING(), fromString("123"), 123.0f)
.fromCase(STRING(), fromString(" 123 "), 123.0f)
@@ -386,9 +386,9 @@ class CastRulesTest {
.fromCase(BOOLEAN(), false, 0.0f),
CastTestSpecBuilder.testCastTo(DOUBLE())
.fromCase(DOUBLE(), null, null)
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("Apache"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("Apache"),
TableRuntimeException.class)
.fromCase(STRING(), fromString("1.234"), 1.234d)
.fromCase(STRING(), fromString("123"), 123.0d)
.fromCase(STRING(), fromString(" 123 "), 123.0d)
@@ -436,8 +436,8 @@ class CastRulesTest {
.fromCase(BOOLEAN(), true, 1.0d)
.fromCase(BOOLEAN(), false, 0.0d),
CastTestSpecBuilder.testCastTo(DATE())
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
.fromCase(
STRING(),
fromString("123"),
@@ -450,7 +450,7 @@ class CastRulesTest {
STRING(),
fromString("2021-09-27 12:34:56.123456789"),
DateTimeUtils.toInternal(LocalDate.of(2021, 9,
27)))
- .fail(STRING(), fromString("2021/09/27"),
TableException.class)
+ .fail(STRING(), fromString("2021/09/27"),
TableRuntimeException.class)
.fromCase(
TIMESTAMP(9),
TIMESTAMP,
@@ -460,8 +460,8 @@ class CastRulesTest {
TIMESTAMP_LTZ,
DateTimeUtils.toInternal(LocalDate.of(2022, 1,
4))),
CastTestSpecBuilder.testCastTo(TIME())
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
.fromCase(
STRING(),
fromString("23"),
@@ -470,8 +470,11 @@ class CastRulesTest {
STRING(),
fromString("23:45"),
DateTimeUtils.toInternal(LocalTime.of(23, 45,
0)))
- .fail(STRING(), fromString("2021-09-27"),
TableException.class)
- .fail(STRING(), fromString("2021-09-27 12:34:56"),
TableException.class)
+ .fail(STRING(), fromString("2021-09-27"),
TableRuntimeException.class)
+ .fail(
+ STRING(),
+ fromString("2021-09-27 12:34:56"),
+ TableRuntimeException.class)
// https://issues.apache.org/jira/browse/FLINK-17224
Currently, fractional
// seconds are lost
.fromCase(
@@ -481,7 +484,7 @@ class CastRulesTest {
.fail(
STRING(),
fromString("2021-09-27 12:34:56.123456789"),
- TableException.class)
+ TableRuntimeException.class)
.fromCase(
TIMESTAMP(6),
TIMESTAMP,
@@ -491,14 +494,14 @@ class CastRulesTest {
TIMESTAMP_LTZ,
DateTimeUtils.toInternal(LocalTime.of(11, 34,
56, 123_000_000))),
CastTestSpecBuilder.testCastTo(TIMESTAMP(9))
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("123"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("123"),
TableRuntimeException.class)
.fromCase(
STRING(),
fromString("2021-09-27"),
timestampDataFromLocalDateTime(2021, 9, 27, 0,
0, 0, 0))
- .fail(STRING(), fromString("2021/09/27"),
TableException.class)
+ .fail(STRING(), fromString("2021/09/27"),
TableRuntimeException.class)
.fromCase(
STRING(),
fromString("2021-09-27 12:34:56.123"),
@@ -578,9 +581,9 @@ class CastRulesTest {
fromString("2021-09-27 12:34:56.12345"),
timestampDataFromLocalDateTime(2021, 9, 27,
12, 34, 56, 123400000)),
CastTestSpecBuilder.testCastTo(TIMESTAMP_LTZ(9))
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("123"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("123"),
TableRuntimeException.class)
.fromCase(
STRING(),
CET_CONTEXT,
@@ -1127,12 +1130,12 @@ class CastRulesTest {
EMPTY_UTF8),
CastTestSpecBuilder.testCastTo(BOOLEAN())
.fromCase(BOOLEAN(), null, null)
- .fail(CHAR(3), fromString("foo"), TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
.fromCase(CHAR(4), fromString("true"), true)
.fromCase(VARCHAR(5), fromString("FalsE"), false)
- .fail(STRING(), fromString("Apache Flink"),
TableException.class)
+ .fail(STRING(), fromString("Apache Flink"),
TableRuntimeException.class)
.fromCase(STRING(), fromString("TRUE"), true)
- .fail(STRING(), fromString(""), TableException.class)
+ .fail(STRING(), fromString(""),
TableRuntimeException.class)
// Should fail when
https://issues.apache.org/jira/browse/FLINK-24576 is
// fixed
.fromCase(
@@ -1253,9 +1256,9 @@ class CastRulesTest {
fromString("Apache"),
new byte[] {65, 112, 97, 99, 104, 101}),
CastTestSpecBuilder.testCastTo(DECIMAL(5, 3))
- .fail(CHAR(3), fromString("foo"), TableException.class)
- .fail(VARCHAR(5), fromString("Flink"),
TableException.class)
- .fail(STRING(), fromString("Apache"),
TableException.class)
+ .fail(CHAR(3), fromString("foo"),
TableRuntimeException.class)
+ .fail(VARCHAR(5), fromString("Flink"),
TableRuntimeException.class)
+ .fail(STRING(), fromString("Apache"),
TableRuntimeException.class)
.fromCase(
STRING(),
fromString("1.234"),
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
index 2d7dec14288..42e6abf3e48 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlJsonUtils.java
@@ -23,8 +23,7 @@ import org.apache.flink.table.api.JsonExistsOnError;
import org.apache.flink.table.api.JsonQueryOnEmptyOrError;
import org.apache.flink.table.api.JsonQueryWrapper;
import org.apache.flink.table.api.JsonValueOnEmptyOrError;
-import org.apache.flink.table.api.TableException;
-import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.table.api.TableRuntimeException;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonValue;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
@@ -103,7 +102,8 @@ public class SqlJsonUtils {
final Object convertedNode = MAPPER.treeToValue(node,
Object.class);
return MAPPER.writeValueAsString(convertedNode);
} catch (JsonProcessingException e) {
- throw new TableException("JSON object could not be serialized: " +
node.asText(), e);
+ throw new TableRuntimeException(
+ "JSON object could not be serialized: " + node.asText(),
e);
}
}
@@ -389,82 +389,82 @@ public class SqlJsonUtils {
}
}
- private static RuntimeException toUnchecked(Exception e) {
- if (e instanceof RuntimeException) {
- return (RuntimeException) e;
+ private static TableRuntimeException toUnchecked(Exception e) {
+ if (e instanceof TableRuntimeException) {
+ return (TableRuntimeException) e;
}
- return new RuntimeException(e);
+ return new TableRuntimeException(e.getMessage(), e);
}
private static RuntimeException illegalJsonPathModeInPathSpec(
String pathMode, String pathSpec) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal jsonpath mode ''%s'' in jsonpath spec:
''%s''",
pathMode, pathSpec));
}
private static RuntimeException illegalJsonPathMode(String pathMode) {
- return new FlinkRuntimeException(String.format("Illegal jsonpath mode
''%s''", pathMode));
+ return new TableRuntimeException(String.format("Illegal jsonpath mode
''%s''", pathMode));
}
private static RuntimeException illegalJsonPathSpec(String pathSpec) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal jsonpath spec ''%s'', format of the spec
should be: ''<lax|strict> $'{'expr'}'''",
pathSpec));
}
private static RuntimeException strictPathModeRequiresNonEmptyValue() {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
"Strict jsonpath mode requires a non empty returned value, but
is null");
}
private static RuntimeException
illegalErrorBehaviorInJsonExistsFunc(String errorBehavior) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal error behavior ''{0}'' specified in
JSON_EXISTS function",
errorBehavior));
}
private static RuntimeException emptyResultOfJsonValueFuncNotAllowed() {
- return new FlinkRuntimeException("Empty result of JSON_VALUE function
is not allowed");
+ return new TableRuntimeException("Empty result of JSON_VALUE function
is not allowed");
}
private static RuntimeException illegalEmptyBehaviorInJsonValueFunc(String
emptyBehavior) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal empty behavior ''{0}'' specified in
JSON_VALUE function",
emptyBehavior));
}
private static RuntimeException illegalErrorBehaviorInJsonValueFunc(String
errorBehavior) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal error behavior ''%s'' specified in JSON_VALUE
function",
errorBehavior));
}
private static RuntimeException
scalarValueRequiredInStrictModeOfJsonValueFunc(String value) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Strict jsonpath mode requires scalar value, and the
actual value is: ''%s''",
value));
}
private static RuntimeException
illegalWrapperBehaviorInJsonQueryFunc(String wrapperBehavior) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal wrapper behavior ''%s'' specified in
JSON_QUERY function",
wrapperBehavior));
}
private static RuntimeException emptyResultOfJsonQueryFuncNotAllowed() {
- return new FlinkRuntimeException("Empty result of JSON_QUERY function
is not allowed");
+ return new TableRuntimeException("Empty result of JSON_QUERY function
is not allowed");
}
private static RuntimeException illegalEmptyBehaviorInJsonQueryFunc(String
emptyBehavior) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal empty behavior ''%s'' specified in JSON_VALUE
function",
emptyBehavior));
@@ -472,14 +472,14 @@ public class SqlJsonUtils {
private static RuntimeException
arrayOrObjectValueRequiredInStrictModeOfJsonQueryFunc(
String value) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Strict jsonpath mode requires array or object value,
and the actual value is: ''%s''",
value));
}
private static RuntimeException illegalErrorBehaviorInJsonQueryFunc(String
errorBehavior) {
- return new FlinkRuntimeException(
+ return new TableRuntimeException(
String.format(
"Illegal error behavior ''%s'' specified in JSON_VALUE
function",
errorBehavior));
