twalthr commented on a change in pull request #18701:
URL: https://github.com/apache/flink/pull/18701#discussion_r803676773
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/StreamPlanner.scala
##########
@@ -129,14 +134,48 @@ class StreamPlanner(
new StreamPlanner(executor, config, moduleManager, functionCatalog,
catalogManager)
}
+ override def loadPlan(planReference: PlanReference): CompiledPlanInternal = {
+ val ctx = createSerdeContext
+ val objectReader: ObjectReader = JsonSerdeUtil.createObjectReader(ctx)
+ val execNodeGraph = planReference match {
+ case filePlanReference: FilePlanReference =>
+ objectReader.readValue(filePlanReference.getFile,
classOf[ExecNodeGraph])
+ case contentPlanReference: ContentPlanReference =>
+ objectReader.readValue(contentPlanReference.getContent,
classOf[ExecNodeGraph])
+ case resourcePlanReference: ResourcePlanReference => {
+ val url = resourcePlanReference.getClassLoader
+ .getResource(resourcePlanReference.getResourcePath)
+ if (url == null) {
+ throw new IOException(
+ "Cannot load the plan reference from classpath: " + planReference);
+ }
+ objectReader.readValue(new File(url.toURI), classOf[ExecNodeGraph])
+ }
+ case _ => throw new IllegalStateException(
+ "Unknown PlanReference. This is a bug, please contact the developers")
+ }
+
+ new ExecNodeGraphCompiledPlan(
+ this,
+ JsonSerdeUtil.createObjectWriter(createSerdeContext)
+ .withDefaultPrettyPrinter()
+ .writeValueAsString(execNodeGraph),
+ execNodeGraph)
+ }
+
override def compilePlan(modifyOperations: util.List[ModifyOperation]):
CompiledPlanInternal = {
validateAndOverrideConfiguration()
val relNodes = modifyOperations.map(translateToRel)
val optimizedRelNodes = optimize(relNodes)
val execGraph = translateToExecNodeGraph(optimizedRelNodes)
cleanupInternalConfigurations()
- new ExecNodeGraphCompiledPlan(createSerdeContext, execGraph)
+ new ExecNodeGraphCompiledPlan(
+ this,
+ JsonSerdeUtil.createObjectWriter(createSerdeContext)
+ .withDefaultPrettyPrinter()
Review comment:
fine with me, we can still introduce a compressed format in the future
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
