You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yu...@apache.org on 2023/08/18 02:20:56 UTC
[flink] branch master updated (5bf5003f5c7 -> 18b4759cc34)
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 5bf5003f5c7 [FLINK-32834] Use descriptive output file names
new 13cb456eb30 [Flink-32356][doc] Add doc for procedure
new 18b4759cc34 [Flink-32356][doc] Add doc for call statement This closes #23130
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content.zh/docs/dev/table/procedures.md | 541 ++++++++++++++++++++++++++
docs/content.zh/docs/dev/table/sql/call.md | 118 ++++++
docs/content/docs/dev/table/procedures.md | 544 +++++++++++++++++++++++++++
docs/content/docs/dev/table/sql/call.md | 118 ++++++
4 files changed, 1321 insertions(+)
create mode 100644 docs/content.zh/docs/dev/table/procedures.md
create mode 100644 docs/content.zh/docs/dev/table/sql/call.md
create mode 100644 docs/content/docs/dev/table/procedures.md
create mode 100644 docs/content/docs/dev/table/sql/call.md
[flink] 01/02: [Flink-32356][doc] Add doc for procedure
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 13cb456eb30dc50dbd47885b7471cd02cfe56cdf
Author: luoyuxia <lu...@alumni.sjtu.edu.cn>
AuthorDate: Wed Aug 2 21:01:57 2023 +0800
[Flink-32356][doc] Add doc for procedure
---
docs/content.zh/docs/dev/table/procedures.md | 541 ++++++++++++++++++++++++++
docs/content/docs/dev/table/procedures.md | 544 +++++++++++++++++++++++++++
2 files changed, 1085 insertions(+)
diff --git a/docs/content.zh/docs/dev/table/procedures.md b/docs/content.zh/docs/dev/table/procedures.md
new file mode 100644
index 00000000000..cb70d759ebc
--- /dev/null
+++ b/docs/content.zh/docs/dev/table/procedures.md
@@ -0,0 +1,541 @@
+---
+title: "存储过程"
+is_beta: true
+weight: 50
+type: docs
+aliases:
+ - /zh/dev/table/procedures.html
+---
+<!--
+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.
+-->
+
+# 存储过程
+
+
+Flink 允许用户在 Table API 和 SQL 中调用存储过程来完成一些特定任务，比如处理数据，数据管理类任务等。存储过程可以通过 `StreamExecutionEnvironment` 来运行 Flink 作业，这使得存储过程更加强大和灵活。
+
+## 开发指南
+------------------
+
+为了调用一个存储过程，需要确保一个 Catalog 可以提供这个存储过程。为了让一个 Catalog 提供存储过程，你首先需要实现一个存储过程，然后在方法 `Catalog.getProcedure(ObjectPath procedurePath)` 返回这个存储过程。
+下面的步骤将展示如何实现一个存储过程并让一个 Catalog 提供这个存储过程。
+
+### 存储过程类
+
+存储过程的实现类必须实现接口 `org.apache.flink.table.procedures.Procedure`。
+
+该实现类必须声明为 `public`， 而不是 `abstract`， 并且可以被全局访问。不允许使用非静态内部类或匿名类。
+
+### Call 方法
+
+存储过程的接口不提供任何方法，存储过程的实现类必须有名为 `call` 的方法，在该方法里面可以实现存储过程实际的逻辑。`call` 方法必须被声明为 `public`， 并且带有一组定义明确的参数。
+
+请注意:
+
+* `call` 方法的第一个参数总是应该为 `ProcedureContext`，该参数提供了方法 `getExecutionEnvironment()` 来得到当前的 `StreamExecutionEnvironment`。通过 `StreamExecutionEnvironment` 可以运行一个 Flink 作业；
+* `call` 方法的返回类型应该永远都是一个数组类型，比如 `int[]`，`String[]`，等等；
+
+更多的细节请参考类 `org.apache.flink.table.procedures.Procedure` 的 Java 文档。
+
+常规的 JVM 方法调用语义是适用的，因此可以：
+- 实现重载的方法，例如 `call(ProcedureContext, Integer)` and `call(ProcedureContext, LocalDateTime)`；
+- 使用变长参数，例如 `call(ProcedureContext, Integer...)`；
+- 使用对象继承，例如 `call(ProcedureContext, Object)` 可接受 `LocalDateTime` 和 `Integer` 作为参数；
+- 也可组合使用，例如 `call(ProcedureContext, Object...)` 可接受所有类型的参数；
+
+如果你希望用 Scala 来实现一个存储过程，对应可变长参数的情况，请添加 `scala.annotation.varargs`。另外，推荐使用装箱的基本类型（比如，使用 `java.lang.Integer` 而不是 `Int`）来支持 `NULL`。
+
+下面的代码片段展示来一个重载存储过程的例子：
+
+{{< tabs "7c5a5392-30d7-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+
+```java
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+
+// 有多个重载 call 方法的存储过程
+public class GenerateSequenceProcedure implements Procedure {
+
+ public long[] call(ProcedureContext context, int n) {
+ return generate(context.getExecutionEnvironment(), n);
+ }
+
+ public long[] call(ProcedureContext context, String n) {
+ return generate(context.getExecutionEnvironment(), Integer.parseInt(n));
+ }
+
+ private long[] generate(StreamExecutionEnvironment env, int n) throws Exception {
+ long[] sequenceN = new long[n];
+ int i = 0;
+ try (CloseableIterator<Long> result = env.fromSequence(0, n - 1).executeAndCollect()) {
+ while (result.hasNext()) {
+ sequenceN[i++] = result.next();
+ }
+ }
+ return sequenceN;
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+import scala.annotation.varargs
+
+// 有多个重载 call 方法的存储过程
+class GenerateSequenceProcedure extends Procedure {
+
+ def call(context: ProcedureContext, a: Integer, b: Integer): Array[Integer] = {
+ Array(a + b)
+ }
+
+ def call(context: ProcedureContext, a: String, b: String): Array[Integer] = {
+ Array(Integer.valueOf(a) + Integer.valueOf(b))
+ }
+
+ @varargs // 类似 Java 的变长参数
+ def call(context: ProcedureContext, d: Double*): Array[Integer] = {
+ Array(d.sum.toInt)
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+### 类型推导
+Table（类似于 SQL 标准）是一种强类型的 API。 因此，存储过程的参数和返回类型都必须映射到 [data type]({{< ref "docs/dev/table/types" >}})。
+
+从逻辑角度看，Planner 需要知道数据类型、精度和小数位数；从 JVM 角度来看，Planner 在调用存储过程时需要知道如何将内部数据结构表示为 JVM 对象。
+
+术语 _类型推导_ 概括了意在验证输入值、推导出参数/返回值数据类型的逻辑。
+
+Flink 存储过程实现了自动的类型推导提取，通过反射从存储过程的类及其 `call` 方法中推导数据类型。如果这种隐式的反射提取方法不成功，则可以通过使用 `@DataTypeHint` 和 `@ProcedureHint` 注解相关参数、类或方法来支持提取存储过程的参数和返回类型，下面展示了有关如何注解存储过程的例子。
+
+需要注意的是虽然存储过程的 `call` 方法必须返回数组类型 `T[]`，但是如果用 `@DataTypeHint` 来注解返回类型，实际上注解的是该数组的元素的类型，即 `T`。
+
+#### 自动类型推导
+
+自动类型推导会检查存储过程的类和 `call` 方法，推导出存储过程参数和结果的数据类型， `@DataTypeHint` 和 `@ProcedurenHint` 注解支持自动类型推导。
+
+有关可以隐式映射到数据类型的类的完整列表, 请参阅[data type extraction section]({{< ref "docs/dev/table/types" >}}#data-type-extraction)。
+
+**`@DataTypeHint`**
+
+在许多情况下，需要支持以 _内联_ 方式自动提取出存储过程参数、返回值的类型。
+
+以下例子展示了如何使用 `@DataTypeHint`，详情可参考该注解类的文档。
+
+{{< tabs "81b297da-30d9-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.InputGroup;
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure;
+import org.apache.flink.types.Row;
+
+// 有多个重载 call 方法的存储过程
+public static class OverloadedProcedure implements Procedure {
+
+ // 不需要 hint
+ public Long[] call(ProcedureContext context, long a, long b) {
+ return new Long[] {a + b};
+ }
+
+ // 定义 decimal 的精度和小数位
+ public @DataTypeHint("DECIMAL(12, 3)") BigDecimal[] call(ProcedureContext context, double a, double b) {
+ return new BigDecimal[] {BigDecimal.valueOf(a + b)};
+ }
+
+ // 定义嵌套数据类型
+ @DataTypeHint("ROW<s STRING, t TIMESTAMP_LTZ(3)>")
+ public Row[] call(ProcedureContext context, int i) {
+ return new Row[] {Row.of(String.valueOf(i), Instant.ofEpochSecond(i))};
+ }
+
+ // 允许任意类型的输入，并输出序列化定制后的值
+ @DataTypeHint(value = "RAW", bridgedTo = ByteBuffer.class)
+ public ByteBuffer[] call(ProcedureContext context, @DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
+ return new ByteBuffer[] {MyUtils.serializeToByteBuffer(o)};
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+import org.apache.flink.table.annotation.DataTypeHint
+import org.apache.flink.table.annotation.InputGroup
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.types.Row
+import scala.annotation.varargs
+
+import org.apache.flink.table.annotation.DataTypeHint
+import org.apache.flink.table.annotation.InputGroup
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+import org.apache.flink.types.Row
+
+// 有多个重载 call 方法的存储过程
+class OverloadedProcedure extends Procedure {
+
+ // 不需要 hint
+ def call(context: ProcedureContext, a: Long, b: Long): Array[Long] = {
+ Array(a + b)
+ }
+
+ // 定义 decimal 的精度和小数位
+ @DataTypeHint("DECIMAL(12, 3)")
+ def call(context: ProcedureContext, a: Double, b: Double): Array[BigDecimal] = {
+ Array(BigDecimal.valueOf(a + b))
+ }
+
+ // 定义嵌套数据类型
+ @DataTypeHint("ROW<s STRING, t TIMESTAMP_LTZ(3)>")
+ def call(context: ProcedureContext, i: Integer): Array[Row] = {
+ Row.of(java.lang.String.valueOf(i), java.time.Instant.ofEpochSecond(i))
+ }
+
+ // 允许任意类型的输入，并输出序列化定制后的值
+ @DataTypeHint(value = "RAW", bridgedTo = classOf[java.nio.ByteBuffer])
+ def call(context: ProcedureContext, @DataTypeHint(inputGroup = InputGroup.ANY) o: Object): Array[java.nio.ByteBuffer] = {
+ Array[MyUtils.serializeToByteBuffer(o)]
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+**`@ProcedureHint`**
+
+有时我们希望一种 `call` 方法可以同时处理多种数据类型，有时又要求对重载的多个 `call` 方法仅声明一次通用的返回类型。
+
+`@ProcedureHint` 注解可以提供从入参数据类型到返回数据类型的映射，它可以在整个存储过程类或 `call` 方法上注解输入和返回的数据类型。可以在类顶部声明一个或多个注解，也可以为类的所有 `call` 方法分别声明一个或多个注解。所有的 hint 参数都是可选的，如果未定义参数，则使用默认的基于反射的类型提取。在函数类顶部定义的 hint 参数被所有 `call` 方法继承。
+
+以下例子展示了如何使用 `@ProcedureHint`，详情可参考该注解类的文档。
+
+{{< tabs "5d205654-30da-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+import org.apache.flink.types.Row;
+
+// 为存储过程类的所有 call 方法指定同一个返回类型
+@ProcedureHint(output = @DataTypeHint("ROW<s STRING, i INT>"))
+public static class OverloadedProcedure implements Procedure {
+
+ public Row[] call(ProcedureContext context, int a, int b) {
+ return new Row[] {Row.of("Sum", a + b)};
+ }
+
+ // 仍然可以重载 call 方法
+ public Row[] call(ProcedureContext context) {
+ return new Row[] {Row.of("Empty args", -1)};
+ }
+}
+
+// 解耦类型推导与 call 方法，类型推导完全取决于 ProcedureHint
+@ProcedureHint(
+ input = {@DataTypeHint("INT"), @DataTypeHint("INT")},
+ output = @DataTypeHint("INT")
+)
+@ProcedureHint(
+ input = {@DataTypeHint("BIGINT"), @DataTypeHint("BIGINT")},
+ output = @DataTypeHint("BIGINT")
+)
+@ProcedureHint(
+ input = {},
+ output = @DataTypeHint("BOOLEAN")
+)
+public static class OverloadedProcedure implements Procedure {
+
+ // 一个 call 方法的实现，确保 call 方法存在于存储过程类中，可以被 JVM 调用
+ public Object[] call(ProcedureContext context, Object... o) {
+ if (o.length == 0) {
+ return new Object[] {false};
+ }
+ return new Object[] {o[0]};
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+
+import org.apache.flink.table.annotation.DataTypeHint
+import org.apache.flink.table.annotation.ProcedureHint
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+import org.apache.flink.types.Row
+import scala.annotation.varargs
+
+// 为存储过程类的所有 call 方法指定同一个返回类型
+@ProcedureHint(output = new DataTypeHint("ROW<s STRING, i INT>"))
+class OverloadedFunction extends Procedure {
+
+ def call(context: ProcedureContext, a: Int, b: Int): Array[Row] = {
+ Array(Row.of("Sum", Int.box(a + b)))
+ }
+
+ // 仍然可以重载 call 方法
+ def call(context: ProcedureContext): Array[Row] = {
+ Array(Row.of("Empty args", Int.box(-1)))
+ }
+}
+
+// 解耦类型推导与 call 方法，类型推导完全取决于 ProcedureHint
+@ProcedureHint(
+ input = Array(new DataTypeHint("INT"), new DataTypeHint("INT")),
+ output = new DataTypeHint("INT")
+)
+@ProcedureHint(
+ input = Array(new DataTypeHint("BIGINT"), new DataTypeHint("BIGINT")),
+ output = new DataTypeHint("BIGINT")
+)
+@ProcedureHint(
+ input = Array(),
+ output = new DataTypeHint("BOOLEAN")
+)
+class OverloadedProcedure extends Procedure {
+
+ // 一个 call 方法的实现，确保 call 方法存在于存储过程类中，可以被 JVM 调用
+ @varargs
+ def call(context: ProcedureContext, o: AnyRef*): Array[AnyRef]= {
+ if (o.length == 0) {
+ Array(Boolean.box(false))
+ }
+ Array(o(0))
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+
+### 在 Catalog 中返回存储过程
+在实现了一个存储过程后，Catalog 可以通过方法 `Catalog.getProcedure(ObjectPath procedurePath)` 来返回该存储过程，下面的例子展示了如何在 Catalog 中返回存储过程。
+另外也可以在 `Catalog.listProcedures(String dbName)` 方法中列出所有的存储过程。
+
+{{< tabs "2ee21ac4-30db-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+
+import java.util.HashMap;
+import java.util.Map;
+
+// 有内置 procedure 的 Catalog
+public class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog {
+
+ static {
+ PROCEDURE_MAP.put(ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure());
+ }
+
+ public CatalogWithBuiltInProcedure(String name) {
+ super(name);
+ }
+
+ @Override
+ public List<String> listProcedures(String dbName) throws DatabaseNotExistException, CatalogException {
+ return PROCEDURE_MAP.keySet().stream().filter(procedurePath -> procedurePath.getDatabaseName().equals(dbName))
+ .map(ObjectPath::getObjectName).collect(Collectors.toList());
+ }
+
+ @Override
+ public Procedure getProcedure(ObjectPath procedurePath) throws ProcedureNotExistException, CatalogException {
+ if (PROCEDURE_MAP.containsKey(procedurePath)) {
+ return PROCEDURE_MAP.get(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName(), procedurePath);
+ }
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
+import org.apache.flink.table.procedures.Procedure;
+
+// 有内置 procedure 的 Catalog
+class CatalogWithBuiltInProcedure(name: String) extends GenericInMemoryCatalog(name) {
+
+ val PROCEDURE_MAP = collection.immutable.HashMap[ObjectPath, Procedure](
+ ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure());
+
+ @throws(classOf[DatabaseNotExistException])
+ @throws(classOf[CatalogException])
+ override def listProcedures(dbName: String): List[String] = {
+ if (!databaseExists(dbName)) {
+ throw new DatabaseNotExistException(getName, dbName);
+ }
+ PROCEDURE_MAP.keySet.filter(procedurePath => procedurePath.getDatabaseName.equals(dbName))
+ .map(procedurePath => procedurePath.getObjectName).toList
+ }
+
+ @throws(classOf[ProcedureNotExistException])
+ override def getProcedure(procedurePath: ObjectPath): Procedure = {
+ if (PROCEDURE_MAP.contains(procedurePath)) {
+ PROCEDURE_MAP(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName, procedurePath)
+ }
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+
+## 例子
+
+下面的例子展示了如何在一个 Catalog 中提供一个存储过程并且通过 `CALL` 语句来调用这个存储过程。详情可参考[开发指南](#implementation-guide)。
+
+{{< tabs "c3edd888-30db-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+
+// 首先实现一个存储过程
+public static class GenerateSequenceProcedure implements Procedure {
+
+ public long[] call(ProcedureContext context, int n) {
+ long[] sequenceN = new long[n];
+ int i = 0;
+ try (CloseableIterator<Long> result = env.fromSequence(0, n - 1).executeAndCollect()) {
+ while (result.hasNext()) {
+ sequenceN[i++] = result.next();
+ }
+ }
+ return sequenceN;
+ }
+}
+
+// 自定义一个 Catalog，并返回该存储过程
+public static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog {
+
+ static {
+ PROCEDURE_MAP.put(ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure());
+ }
+ // 省略一些方法
+ // ...
+ @Override
+ public Procedure getProcedure(ObjectPath procedurePath) throws ProcedureNotExistException, CatalogException {
+ if (PROCEDURE_MAP.containsKey(procedurePath)) {
+ return PROCEDURE_MAP.get(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName(), procedurePath);
+ }
+ }
+}
+
+TableEnvironment tEnv = TableEnvironment.create(...);
+// 注册这个 Catalog
+tEnv.registerCatalog("my_catalog", new CatalogWithBuiltInProcedure());
+// 通过 Call 语句调用该存储过程
+tEnv.executeSql("call my_catalog.`system`.generate_n(5)");
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+
+import org.apache.flink.table.catalog.{GenericInMemoryCatalog, ObjectPath}
+import org.apache.flink.table.catalog.exceptions.{CatalogException, ProcedureNotExistException}
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+
+// 首先实现一个存储过程
+class GenerateSequenceProcedure extends Procedure {
+
+ def call(context: ProcedureContext, n: Integer): Array[Long] = {
+ val env = context.getExecutionEnvironment
+ val sequenceN = Array[Long]
+ var i = 0;
+ env.fromSequence(0, n - 1).executeAndCollect()
+ .forEachRemaining(r => {
+ sequenceN(i) = r
+ i = i + 1
+ })
+ sequenceN;
+ }
+}
+
+// 然后在一个自定义的 catalog 返回该 procedure
+class CatalogWithBuiltInProcedure(name: String) extends GenericInMemoryCatalog(name) {
+
+ val PROCEDURE_MAP = collection.immutable.HashMap[ObjectPath, Procedure](ObjectPath.fromString("system.generate_n"),
+ new GenerateSequenceProcedure());
+
+ // 省略一些方法
+ // ...
+
+ @throws(classOf[ProcedureNotExistException])
+ override def getProcedure(procedurePath: ObjectPath): Procedure = {
+ if (PROCEDURE_MAP.contains(procedurePath)) {
+ PROCEDURE_MAP(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName, procedurePath)
+ }
+ }
+}
+
+TableEnvironment tEnv = TableEnvironment.create(...)
+// 注册该 catalog
+tEnv.registerCatalog("my_catalog", new CatalogWithBuiltInProcedure())
+// 通过 Call 语句调用该存储过程
+tEnv.executeSql("call my_catalog.`system`.generate_n(5)")
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+
diff --git a/docs/content/docs/dev/table/procedures.md b/docs/content/docs/dev/table/procedures.md
new file mode 100644
index 00000000000..0122a0de191
--- /dev/null
+++ b/docs/content/docs/dev/table/procedures.md
@@ -0,0 +1,544 @@
+---
+title: "Procedures"
+is_beta: true
+weight: 50
+type: docs
+aliases:
+ - /dev/table/procedures.html
+---
+<!--
+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.
+-->
+
+# Procedures
+
+Flink Table API & SQL empowers users to perform data manipulation and administrative tasks with procedures. Procedures can run FLINK jobs with the provided `StreamExecutionEnvironment`, making them more powerful and flexible.
+
+## Implementation Guide
+
+To call a procedure, it must be available in a catalog. To provide procedures in a catalog, you need to implement the procedure and then return it using the `Catalog.getProcedure(ObjectPath procedurePath)` method.
+The following steps will guild you on how to implement and provide a procedure in a catalog.
+
+### Procedure Class
+
+An implementation class must implement the interface `org.apache.flink.table.procedures.Procedure`.
+
+The class must be declared `public`, not `abstract`, and should be globally accessible. Thus, non-static inner or anonymous classes are not allowed.
+
+### Call Methods
+
+The interface doesn't provide any method，you have to define a method named `call` in which you can implement the logic of the procedure.
+The methods must be declared `public` and take a well-defined set of arguments.
+
+Please note:
+
+* The first parameter of the method `call` should always be `ProcedureContext` which provides the method `getExecutionEnvironment` to get a `StreamExecutionEnvironment` for running a Flink Job
+* The return type should always be an array, like `int[]`, `String[]`, etc
+
+More detail can be found in the Java doc of the class `org.apache.flink.table.procedures.Procedure`.
+
+Regular JVM method calling semantics apply. Therefore, it is possible to:
+- implement overloaded methods such as `call(ProcedureContext, Integer)` and `call(ProcedureContext, LocalDateTime)`
+- use var-args such as `call(ProcedureContext, Integer...)`
+- use object inheritance such as `call(ProcedureContext, Object)` that takes both `LocalDateTime` and `Integer`
+- and combinations of the above such as `call(ProcedureContext, Object...)` that takes all kinds of arguments
+
+If you intend to implement procedures in Scala, please add the `scala.annotation.varargs` annotation in
+case of variable arguments. Furthermore, it is recommended to use boxed primitives (e.g. `java.lang.Integer`
+instead of `Int`) to support `NULL`.
+
+The following snippets shows an example of an overloaded procedure:
+
+{{< tabs "0819d780-3052-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+
+```java
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+
+// procedure with overloaded call methods
+public class GenerateSequenceProcedure implements Procedure {
+
+ public long[] call(ProcedureContext context, int n) {
+ return generate(context.getExecutionEnvironment(), n);
+ }
+
+ public long[] call(ProcedureContext context, String n) {
+ return generate(context.getExecutionEnvironment(), Integer.parseInt(n));
+ }
+
+ private long[] generate(StreamExecutionEnvironment env, int n) throws Exception {
+ long[] sequenceN = new long[n];
+ int i = 0;
+ try (CloseableIterator<Long> result = env.fromSequence(0, n - 1).executeAndCollect()) {
+ while (result.hasNext()) {
+ sequenceN[i++] = result.next();
+ }
+ }
+ return sequenceN;
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+import scala.annotation.varargs
+
+// procedures with overloaded call methods
+class GenerateSequenceProcedure extends Procedure {
+
+ def call(context: ProcedureContext, a: Integer, b: Integer): Array[Integer] = {
+ Array(a + b)
+ }
+
+ def call(context: ProcedureContext, a: String, b: String): Array[Integer] = {
+ Array(Integer.valueOf(a) + Integer.valueOf(b))
+ }
+
+ @varargs // generate var-args like Java
+ def call(context: ProcedureContext, d: Double*): Array[Integer] = {
+ Array(d.sum.toInt)
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+### Type Inference
+The table ecosystem (similar to the SQL standard) is a strongly typed API. Therefore, both procedure parameters and return types must be mapped to a [data type]({{< ref "docs/dev/table/types" >}}).
+
+From a logical perspective, the planner needs information about expected types, precision, and scale. From a JVM perspective, the planner needs information about how internal data structures are represented as JVM objects when calling a procedure.
+
+The logic for validating input arguments and deriving data types for both the parameters and the result of a procedure is summarized under the term _type inference_.
+
+Flink's procedures implement an automatic type inference extraction that derives data types from the procedure's class and its `call` methods via reflection. If this implicit reflective extraction approach is not successful, the extraction process can be supported by annotating affected parameters, classes, or methods with `@DataTypeHint` and `@ProcedureHint`. More examples on how to annotate procedures are shown below.
+
+Note: although the return type in `call` method must be array type `T[]`, if use `@DataTypeHint` to annotate the return type, it's actually expected to annotate the component type of the array type, which is actually `T`.
+
+#### Automatic Type Inference
+
+The automatic type inference inspects the procedure's class and `call` methods to derive data types for the arguments and result of a procedure. `@DataTypeHint` and `@ProcedureHint` annotations support the automatic extraction.
+
+For a full list of classes that can be implicitly mapped to a data type, please refer to the [data type extraction section]({{< ref "docs/dev/table/types" >}}#data-type-extraction).
+
+**`@DataTypeHint`**
+
+In many scenarios, it is required to support the automatic extraction _inline_ for parameters and return types of a procedure
+
+The following example shows how to use data type hints. More information can be found in the documentation of the annotation class.
+
+{{< tabs "9414057c-3051-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.InputGroup;
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure;
+import org.apache.flink.types.Row;
+
+// procedure with overloaded call methods
+public static class OverloadedProcedure implements Procedure {
+
+ // no hint required
+ public Long[] call(ProcedureContext context, long a, long b) {
+ return new Long[] {a + b};
+ }
+
+ // define the precision and scale of a decimal
+ public @DataTypeHint("DECIMAL(12, 3)") BigDecimal[] call(ProcedureContext context, double a, double b) {
+ return new BigDecimal[] {BigDecimal.valueOf(a + b)};
+ }
+
+ // define a nested data type
+ @DataTypeHint("ROW<s STRING, t TIMESTAMP_LTZ(3)>")
+ public Row[] call(ProcedureContext context, int i) {
+ return new Row[] {Row.of(String.valueOf(i), Instant.ofEpochSecond(i))};
+ }
+
+ // allow wildcard input and custom serialized output
+ @DataTypeHint(value = "RAW", bridgedTo = ByteBuffer.class)
+ public ByteBuffer[] call(ProcedureContext context, @DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
+ return new ByteBuffer[] {MyUtils.serializeToByteBuffer(o)};
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+import org.apache.flink.table.annotation.DataTypeHint
+import org.apache.flink.table.annotation.InputGroup
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+import org.apache.flink.types.Row
+
+// procedure with overloaded call methods
+class OverloadedProcedure extends Procedure {
+
+ // no hint required
+ def call(context: ProcedureContext, a: Long, b: Long): Array[Long] = {
+ Array(a + b)
+ }
+
+ // define the precision and scale of a decimal
+ @DataTypeHint("DECIMAL(12, 3)")
+ def call(context: ProcedureContext, a: Double, b: Double): Array[BigDecimal] = {
+ Array(BigDecimal.valueOf(a + b))
+ }
+
+ // define a nested data type
+ @DataTypeHint("ROW<s STRING, t TIMESTAMP_LTZ(3)>")
+ def call(context: ProcedureContext, i: Integer): Array[Row] = {
+ Row.of(java.lang.String.valueOf(i), java.time.Instant.ofEpochSecond(i))
+ }
+
+ // allow wildcard input and custom serialized output
+ @DataTypeHint(value = "RAW", bridgedTo = classOf[java.nio.ByteBuffer])
+ def call(context: ProcedureContext, @DataTypeHint(inputGroup = InputGroup.ANY) o: Object): Array[java.nio.ByteBuffer] = {
+ Array[MyUtils.serializeToByteBuffer(o)]
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+**`@ProcedureHint`**
+
+In some scenarios, it is desirable that one `call` method handles multiple different data types at the same time. Furthermore, in some scenarios, overloaded `call` methods have a common result type that should be declared only once.
+
+The `@ProcedureHint` annotation can provide a mapping from argument data types to a result data type. It enables annotating entire procedure classes or `call` methods for input and result data types. One or more annotations can be declared on top of a class or individually for each `call` method for overloading procedure signatures. All hint parameters are optional. If a parameter is not defined, the default reflection-based extraction is used. Hint parameters defined on top of a procedu [...]
+
+The following example shows how to use procedure hints. More information can be found in the documentation of the annotation class.
+
+{{< tabs "16d50628-3052-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+import org.apache.flink.types.Row;
+
+// procedure with overloaded call methods
+// but globally defined output type
+@ProcedureHint(output = @DataTypeHint("ROW<s STRING, i INT>"))
+public static class OverloadedProcedure implements Procedure {
+
+ public Row[] call(ProcedureContext context, int a, int b) {
+ return new Row[] {Row.of("Sum", a + b)};
+ }
+
+ // overloading of arguments is still possible
+ public Row[] call(ProcedureContext context) {
+ return new Row[] {Row.of("Empty args", -1)};
+ }
+}
+
+// decouples the type inference from call methods,
+// the type inference is entirely determined by the procedure hints
+@ProcedureHint(
+ input = {@DataTypeHint("INT"), @DataTypeHint("INT")},
+ output = @DataTypeHint("INT")
+)
+@ProcedureHint(
+ input = {@DataTypeHint("BIGINT"), @DataTypeHint("BIGINT")},
+ output = @DataTypeHint("BIGINT")
+)
+@ProcedureHint(
+ input = {},
+ output = @DataTypeHint("BOOLEAN")
+)
+public static class OverloadedProcedure implements Procedure {
+
+ // an implementer just needs to make sure that a method exists
+ // that can be called by the JVM
+ public Object[] call(ProcedureContext context, Object... o) {
+ if (o.length == 0) {
+ return new Object[] {false};
+ }
+ return new Object[] {o[0]};
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+
+import org.apache.flink.table.annotation.DataTypeHint
+import org.apache.flink.table.annotation.ProcedureHint
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+import org.apache.flink.types.Row
+import scala.annotation.varargs
+
+// procedure with overloaded call methods
+// but globally defined output type
+@ProcedureHint(output = new DataTypeHint("ROW<s STRING, i INT>"))
+class OverloadedFunction extends Procedure {
+
+ def call(context: ProcedureContext, a: Int, b: Int): Array[Row] = {
+ Array(Row.of("Sum", Int.box(a + b)))
+ }
+
+ // overloading of arguments is still possible
+ def call(context: ProcedureContext): Array[Row] = {
+ Array(Row.of("Empty args", Int.box(-1)))
+ }
+}
+
+// decouples the type inference from call methods,
+// the type inference is entirely determined by the function hints
+@ProcedureHint(
+ input = Array(new DataTypeHint("INT"), new DataTypeHint("INT")),
+ output = new DataTypeHint("INT")
+)
+@ProcedureHint(
+ input = Array(new DataTypeHint("BIGINT"), new DataTypeHint("BIGINT")),
+ output = new DataTypeHint("BIGINT")
+)
+@ProcedureHint(
+ input = Array(),
+ output = new DataTypeHint("BOOLEAN")
+)
+class OverloadedProcedure extends Procedure {
+
+ // an implementer just needs to make sure that a method exists
+ // that can be called by the JVM
+ @varargs
+ def call(context: ProcedureContext, o: AnyRef*): Array[AnyRef]= {
+ if (o.length == 0) {
+ Array(Boolean.box(false))
+ }
+ Array(o(0))
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+
+### Return Procedure in Catalog
+After implementing a procedure, the catalog can then return the procedure in method `Catalog.getProcedure(ObjectPath procedurePath)`. The following example shows how to return it in a catalog.
+Also, it's expected to list all the procedures in method `Catalog.listProcedures(String dbName)`.
+
+{{< tabs "2e7d4b78-3052-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+
+import java.util.HashMap;
+import java.util.Map;
+
+// catalog with built-in procedures
+public static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog {
+ static {
+ PROCEDURE_MAP.put(ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure());
+ }
+
+ public CatalogWithBuiltInProcedure(String name) {
+ super(name);
+ }
+
+ @Override
+ public List<String> listProcedures(String dbName) throws DatabaseNotExistException, CatalogException {
+ if (!databaseExists(dbName)) {
+ throw new DatabaseNotExistException(getName(), dbName);
+ }
+ return PROCEDURE_MAP.keySet().stream().filter(procedurePath -> procedurePath.getDatabaseName().equals(dbName))
+ .map(ObjectPath::getObjectName).collect(Collectors.toList());
+ }
+
+ @Override
+ public Procedure getProcedure(ObjectPath procedurePath) throws ProcedureNotExistException, CatalogException {
+ if (PROCEDURE_MAP.containsKey(procedurePath)) {
+ return PROCEDURE_MAP.get(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName(), procedurePath);
+ }
+ }
+}
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
+import org.apache.flink.table.procedures.Procedure;
+
+// catalog with built-in procedures
+class CatalogWithBuiltInProcedure(name: String) extends GenericInMemoryCatalog(name) {
+
+ val PROCEDURE_MAP = collection.immutable.HashMap[ObjectPath, Procedure](
+ ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure());
+
+ @throws(classOf[DatabaseNotExistException])
+ @throws(classOf[CatalogException])
+ override def listProcedures(dbName: String): List[String] = {
+ if (!databaseExists(dbName)) {
+ throw new DatabaseNotExistException(getName, dbName);
+ }
+ PROCEDURE_MAP.keySet.filter(procedurePath => procedurePath.getDatabaseName.equals(dbName))
+ .map(procedurePath => procedurePath.getObjectName).toList
+ }
+
+ @throws(classOf[ProcedureNotExistException])
+ override def getProcedure(procedurePath: ObjectPath): Procedure = {
+ if (PROCEDURE_MAP.contains(procedurePath)) {
+ PROCEDURE_MAP(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName, procedurePath)
+ }
+ }
+}
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+
+## Examples
+
+The following example shows how to provide a procedure in a `Catalog` and call it with `CALL` statement. See the [Implementation Guide](#implementation-guide) for more details.
+
+{{< tabs "1ed81b8a-3052-11ee-be56-0242ac120002" >}}
+{{< tab "Java" >}}
+```java
+
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.table.procedures.Procedure;
+
+// first implement a procedure
+public class GenerateSequenceProcedure implements Procedure {
+
+ public long[] call(ProcedureContext context, int n) {
+ long[] sequenceN = new long[n];
+ int i = 0;
+ try (CloseableIterator<Long> result = env.fromSequence(0, n - 1).executeAndCollect()) {
+ while (result.hasNext()) {
+ sequenceN[i++] = result.next();
+ }
+ }
+ return sequenceN;
+ }
+}
+
+// then provide the procedure in a custom catalog
+public static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog {
+
+ static {
+ PROCEDURE_MAP.put(ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure());
+ }
+ // emit some methods
+ // ...
+ @Override
+ public Procedure getProcedure(ObjectPath procedurePath) throws ProcedureNotExistException, CatalogException {
+ if (PROCEDURE_MAP.containsKey(procedurePath)) {
+ return PROCEDURE_MAP.get(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName(), procedurePath);
+ }
+ }
+}
+
+TableEnvironment tEnv = TableEnvironment.create(...);
+// register the catalog
+tEnv.registerCatalog("my_catalog", new CatalogWithBuiltInProcedure());
+// call the procedure with CALL statement
+tEnv.executeSql("call my_catalog.`system`.generate_n(5)");
+
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+
+import org.apache.flink.table.catalog.{GenericInMemoryCatalog, ObjectPath}
+import org.apache.flink.table.catalog.exceptions.{CatalogException, ProcedureNotExistException}
+import org.apache.flink.table.procedure.ProcedureContext
+import org.apache.flink.table.procedures.Procedure
+
+// first implement a procedure
+class GenerateSequenceProcedure extends Procedure {
+
+ def call(context: ProcedureContext, n: Integer): Array[Long] = {
+ val env = context.getExecutionEnvironment
+ val sequenceN = Array[Long]
+ var i = 0;
+ env.fromSequence(0, n - 1).executeAndCollect()
+ .forEachRemaining(r => {
+ sequenceN(i) = r
+ i = i + 1
+ })
+ sequenceN;
+ }
+}
+
+// then provide the procedure in a custom catalog
+class CatalogWithBuiltInProcedure(name: String) extends GenericInMemoryCatalog(name) {
+
+ val PROCEDURE_MAP = collection.immutable.HashMap[ObjectPath, Procedure](ObjectPath.fromString("system.generate_n"),
+ new GenerateSequenceProcedure());
+
+ // emit some methods
+ // ...
+
+ @throws(classOf[ProcedureNotExistException])
+ override def getProcedure(procedurePath: ObjectPath): Procedure = {
+ if (PROCEDURE_MAP.contains(procedurePath)) {
+ PROCEDURE_MAP(procedurePath);
+ } else {
+ throw new ProcedureNotExistException(getName, procedurePath)
+ }
+ }
+}
+
+TableEnvironment tEnv = TableEnvironment.create(...)
+// register the catalog
+tEnv.registerCatalog("my_catalog", new CatalogWithBuiltInProcedure())
+// call the procedure with CALL statement
+tEnv.executeSql("call my_catalog.`system`.generate_n(5)")
+
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+
[flink] 02/02: [Flink-32356][doc] Add doc for call statement This closes #23130
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 18b4759cc3405e784205f6df55dd7a4f0ee2d3b7
Author: luoyuxia <lu...@alumni.sjtu.edu.cn>
AuthorDate: Wed Aug 2 21:02:06 2023 +0800
[Flink-32356][doc] Add doc for call statement
This closes #23130
---
docs/content.zh/docs/dev/table/sql/call.md | 118 +++++++++++++++++++++++++++++
docs/content/docs/dev/table/sql/call.md | 118 +++++++++++++++++++++++++++++
2 files changed, 236 insertions(+)
diff --git a/docs/content.zh/docs/dev/table/sql/call.md b/docs/content.zh/docs/dev/table/sql/call.md
new file mode 100644
index 00000000000..3899b80b4d5
--- /dev/null
+++ b/docs/content.zh/docs/dev/table/sql/call.md
@@ -0,0 +1,118 @@
+---
+title: "CALL 语句"
+weight: 19
+type: docs
+aliases:
+- /zh/dev/table/sql/call.html
+---
+<!--
+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.
+-->
+
+# Call Statements
+
+`Call` 语句用来调用存储过程。存储过程通常是用来执行一些数据操作和管理任务的。
+
+<span class="label label-danger">注意</span> 目前 `Call` 语句要求被调用的存储过程在对应的 Catalog 中。所以，在调用存储过程前，请确保该存储过程在对应的 Catalog 中。
+否则就会抛出异常。你可能需要查看对应 Catalog 的文档来知道该 Catalog 有哪些存储过程可用。要实现一个存储过程，请参阅 [存储过程]({{< ref "docs/dev/table/procedures" >}})。
+
+## Run a CALL statement
+
+{{< tabs "call statement" >}}
+
+{{< tab "Java" >}}
+
+CALL 语句可以使用 `TableEnvironment` 中的 `executeSql()` 方法执行。`executeSql()` 方法执行 CALL 语句时会立即调用这个存储过程，并且返回一个 `TableResult` 对象，通过该对象可以获取调用存储过程的结果。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 CALL 语句。
+
+{{< /tab >}}
+{{< tab "Scala" >}}
+
+CALL 语句可以使用 `TableEnvironment` 中的 `executeSql()` 方法执行。`executeSql()` 方法执行 CALL 语句时会立即调用这个存储过程，并且返回一个 `TableResult` 对象，通过该对象可以获取调用存储过程的结果。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 CALL 语句。
+{{< /tab >}}
+{{< tab "Python" >}}
+
+CALL 语句可以使用 `TableEnvironment` 中的 `execute_sql()` 方法执行。`execute_sql()` 方法执行 CALL 语句时会立即调用这个存储过程，并且返回一个 `TableResult` 对象，通过该对象可以获取调用存储过程的结果。
+
+以下的例子展示了如何在 `TableEnvironment` 中执行一条 CALL 语句。
+
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+
+可以在 [SQL CLI]({{< ref "docs/dev/table/sqlClient" >}}) 中执行 CALL 语句。
+
+以下的例子展示了如何在 SQL CLI 中执行一条 CALL 语句。
+
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< tabs "da4af028-303d-11ee-be56-0242ac120002" >}}
+
+{{< tab "Java" >}}
+```java
+TableEnvironment tEnv = TableEnvironment.create(...);
+
+// 假设存储过程 `generate_n` 已经存在于当前 catalog 的 `system` 数据库
+tEnv.executeSql("CALL `system`.generate_n(4)").print();
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val tEnv = TableEnvironment.create(...)
+
+// 假设存储过程 `generate_n` 已经存在于当前 catalog 的 `system` 数据库
+tEnv.executeSql("CALL `system`.generate_n(4)").print()
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+table_env = TableEnvironment.create(...)
+
+# 假设存储过程 `generate_n` 已经存在于当前 catalog 的 `system` 数据库
+table_env.execute_sql().print()
+```
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+```sql
+-- 假设存储过程 `generate_n` 已经存在于当前 catalog 的 `system` 数据库
+Flink SQL> CALL `system`.generate_n(4);
++--------+
+| result |
++--------+
+| 0 |
+| 1 |
+| 2 |
+| 3 |
++--------+
+4 rows in set
+!ok
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< top >}}
+
+## Syntax
+
+```sql
+CALL [catalog_name.][database_name.]procedure_name ([ expression [, expression]* ] )
+```
+
diff --git a/docs/content/docs/dev/table/sql/call.md b/docs/content/docs/dev/table/sql/call.md
new file mode 100644
index 00000000000..ca54b7b207f
--- /dev/null
+++ b/docs/content/docs/dev/table/sql/call.md
@@ -0,0 +1,118 @@
+---
+title: "CALL Statements"
+weight: 19
+type: docs
+aliases:
+- /dev/table/sql/call.html
+---
+<!--
+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.
+-->
+
+# Call Statements
+
+`Call` statements are used to call a stored procedure which is usually provided to perform data manipulation or administrative tasks.
+
+<span class="label label-danger">Attention</span> Currently, `Call` statements require the procedure called to exist in the corresponding catalog. So, please make sure the procedure exists in the catalog.
+If it doesn't exist, it'll throw an exception. You may need to refer to the doc of the catalog to see the available procedures. To implement an procedure, please refer to [Procedure]({{< ref "docs/dev/table/procedures" >}}).
+
+## Run a CALL statement
+
+{{< tabs "call statement" >}}
+
+{{< tab "Java" >}}
+
+CALL statements can be executed with the `executeSql()` method of the `TableEnvironment`. The `executeSql()` will immediately call the procedure, and return a `TableResult` instance which associates the procedure.
+
+The following examples show how to execute a CALL statement in `TableEnvironment`.
+
+{{< /tab >}}
+{{< tab "Scala" >}}
+
+CALL statements can be executed with the `executeSql()` method of the `TableEnvironment`. The `executeSql()` will immediately call the procedure, and return a `TableResult` instance which associates the procedure.
+
+The following examples show how to execute a single CALL statement in `TableEnvironment`.
+{{< /tab >}}
+{{< tab "Python" >}}
+
+CALL statements can be executed with the `execute_sql()` method of the `TableEnvironment`. The `executeSql()` will immediately call the procedure, and return a `TableResult` instance which associates the procedure.
+
+The following examples show how to execute a single CALL statement in `TableEnvironment`.
+
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+
+CALL statements can be executed in [SQL CLI]({{< ref "docs/dev/table/sqlClient" >}}).
+
+The following examples show how to execute a CALL statement in SQL CLI.
+
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< tabs "dcaced3c-3036-11ee-be56-0242ac120002" >}}
+
+{{< tab "Java" >}}
+```java
+TableEnvironment tEnv = TableEnvironment.create(...);
+
+// assuming the procedure `generate_n` has existed in `system` database of the current catalog
+tEnv.executeSql("CALL `system`.generate_n(4)").print();
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+```scala
+val tEnv = TableEnvironment.create(...)
+
+// assuming the procedure `generate_n` has existed in `system` database of the current catalog
+tEnv.executeSql("CALL `system`.generate_n(4)").print()
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+table_env = TableEnvironment.create(...)
+
+# assuming the procedure `generate_n` has existed in `system` database of the current catalog
+table_env.execute_sql().print()
+```
+{{< /tab >}}
+{{< tab "SQL CLI" >}}
+```sql
+// assuming the procedure `generate_n` has existed in `system` database of the current catalog
+Flink SQL> CALL `system`.generate_n(4);
++--------+
+| result |
++--------+
+| 0 |
+| 1 |
+| 2 |
+| 3 |
++--------+
+4 rows in set
+!ok
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< top >}}
+
+## Syntax
+
+```sql
+CALL [catalog_name.][database_name.]procedure_name ([ expression [, expression]* ] )
+```
+