[FLINK-18526][python][docs] Add configuration of Python UDF to use Ma…

docs/dev/table/python/python_udfs.md

@@ -48,6 +48,9 @@ class HashCode(ScalarFunction):
 
 table_env = BatchTableEnvironment.create(env)
 
+# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
+table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m')
+
 # register the Python function
 table_env.register_function("hash_code", udf(HashCode(), DataTypes.BIGINT(), DataTypes.BIGINT()))
 
@@ -58,6 +61,8 @@ my_table.select("string, bigint, bigint.hash_code(), hash_code(bigint)")
 table_env.sql_query("SELECT string, bigint, hash_code(bigint) FROM MyTable")
 {% endhighlight %}
 
+<span class="label label-info">Note</span> If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**.
+
 It also supports to use Java/Scala scalar functions in Python Table API programs.
 
 {% highlight python %}
@@ -76,6 +81,9 @@ public class HashCode extends ScalarFunction {
 
 table_env = BatchTableEnvironment.create(env)
 
+# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
+table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m')
+
 # register the Java function
 table_env.register_java_function("hash_code", "my.java.function.HashCode")
 
@@ -86,6 +94,8 @@ my_table.select("string.hash_code(), hash_code(string)")
 table_env.sql_query("SELECT string, bigint, hash_code(string) FROM MyTable")
 {% endhighlight %}
 
+<span class="label label-info">Note</span> If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**.
+
 There are many ways to define a Python scalar function besides extending the base class `ScalarFunction`.
 The following examples show the different ways to define a Python scalar function which takes two columns of
 bigint as the input parameters and returns the sum of them as the result.
@@ -145,6 +155,9 @@ env = StreamExecutionEnvironment.get_execution_environment()
 table_env = StreamTableEnvironment.create(env)
 my_table = ...  # type: Table, table schema: [a: String]
 
+# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
+table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m')
+
 # register the Python Table Function
 table_env.register_function("split", udtf(Split(), DataTypes.STRING(), [DataTypes.STRING(), DataTypes.INT()]))
 
@@ -158,6 +171,8 @@ table_env.sql_query("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE
 
 {% endhighlight %}
 
+<span class="label label-info">Note</span> If not using RocksDB as state backend, you can configure the python worker to use the managed memory of 
+taskmanager by setting **python.fn-execution.memory.managed** to be **true**.
 
 It also supports to use Java/Scala table functions in Python Table API programs.
 {% highlight python %}
@@ -182,6 +197,9 @@ env = StreamExecutionEnvironment.get_execution_environment()
 table_env = StreamTableEnvironment.create(env)
 my_table = ...  # type: Table, table schema: [a: String]
 
+# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
+table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m')
+
 # Register the java function.
 table_env.register_java_function("split", "my.java.function.Split")
 
@@ -198,6 +216,8 @@ table_env.sql_query("SELECT a, word, length FROM MyTable, LATERAL TABLE(split(a)
 table_env.sql_query("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE(split(a)) as T(word, length) ON TRUE")
 {% endhighlight %}
 
+<span class="label label-info">Note</span> If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**.
+
 Like Python scalar functions, you can use the above five ways to define Python TableFunctions.
 
 <span class="label label-info">Note</span> The only difference is that the return type of Python Table Functions needs to be an iterable, iterator or generator.

docs/dev/table/python/vectorized_python_udfs.md

@@ -54,6 +54,9 @@ def add(i, j):
 
 table_env = BatchTableEnvironment.create(env)
 
+# configure the off-heap memory of current taskmanager to enable the python worker uses off-heap memory.
+table_env.get_config().get_configuration().set_boolean("taskmanager.memory.task.off-heap.size", '80m')
+
 # register the vectorized Python scalar function
 table_env.register_function("add", add)
 
@@ -63,3 +66,5 @@ my_table.select("add(bigint, bigint)")
 # use the vectorized Python scalar function in SQL API
 table_env.sql_query("SELECT add(bigint, bigint) FROM MyTable")
 {% endhighlight %}
+
+<span class="label label-info">Note</span> If not using RocksDB as state backend, you can configure the python worker to use the managed memory of taskmanager by setting **python.fn-execution.memory.managed** to be **true**.