Whole-stage Code Generation in Flink