zhuzhurk commented on a change in pull request #10452:
[FLINK-15035][table-planner-blink] Introduce unknown memory setting to table in
blink planner
URL: https://github.com/apache/flink/pull/10452#discussion_r355105514
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/exec/BatchExecNode.scala
##########
@@ -32,4 +33,13 @@ trait BatchExecNode[T] extends ExecNode[BatchPlanner, T]
with Logging {
*/
def getDamBehavior: DamBehavior
+ def setManagedMemoryWeight[X](
+ transformation: Transformation[X], memoryBytes: Long): Transformation[X]
= {
+ // Using Bytes can easily overflow
+ // Using MebiBytes to cast to int
+ // Careful about zero
+ val memoryMB = Math.max(1, (memoryBytes >> 20).toInt)
Review comment:
Just to confirm whether users' custom sources/sinks would use managed memory?
If user source/sink would not use any managed memory, table can also
override user operators' managed memory size/weight to 0. (table can see all
operators in the StreamGraph and knows which operators are from table, the
others should be source/sink operators I think)
If user the source/sink is possible to use managed memory, then table should
guarantee that those user operators would still be able to acquire manage
memory, considering the competition from managed memory weights of table
operators. Otherwise those source/sink tasks may fail.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
