xintongsong commented on a change in pull request #12370:
URL: https://github.com/apache/flink/pull/12370#discussion_r431671864
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonPythonBase.scala
##########
@@ -150,6 +154,35 @@ trait CommonPythonBase {
}
realEnv
}
+
+ private def usingManagedMemory(config: Configuration): Boolean = {
+ val clazz = loadClass("org.apache.flink.python.PythonOptions")
+ config.getBoolean(clazz.getField("USE_MANAGED_MEMORY").get(null)
+ .asInstanceOf[ConfigOption[java.lang.Boolean]])
+ }
+
+ private def getPythonWorkerMemory(config: Configuration): Long = {
+ val clazz = loadClass("org.apache.flink.python.PythonOptions")
+ val pythonFrameworkMemorySize = MemorySize.parse(
+ config.getString(
+ clazz.getField("PYTHON_FRAMEWORK_MEMORY_SIZE").get(null)
+ .asInstanceOf[ConfigOption[String]]))
+ val pythonBufferMemorySize = MemorySize.parse(
+ config.getString(
+ clazz.getField("PYTHON_DATA_BUFFER_MEMORY_SIZE").get(null)
+ .asInstanceOf[ConfigOption[String]]))
+ pythonFrameworkMemorySize.add(pythonBufferMemorySize).getBytes
+ }
+
+ private def setDefaultConfigurations(config: Configuration): Unit = {
+ if (!usingManagedMemory(config) &&
!config.contains(TaskManagerOptions.TASK_OFF_HEAP_MEMORY)) {
+ // Set the default value for the task off-heap memory if not set if the
Python worker isn't
+ // using managed memory. Note that this is best effort attempt and if a
task slot contains
+ // multiple Python workers, users should set the task off-heap memory
manually.
+ config.setString(TaskManagerOptions.TASK_OFF_HEAP_MEMORY.key(),
+ getPythonWorkerMemory(config) + "b")
+ }
Review comment:
I'm not sure about this behavior.
I think this kind of auto magic is usually implicit and hard to understand.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
