[AURON #1833] Refactor from_proto.rs to planner.rs

Cargo.toml

@@ -23,7 +23,7 @@ members = [
     "native-engine/datafusion-ext-plans",
     "native-engine/auron",
     "native-engine/auron-jni-bridge",
-    "native-engine/auron-serde",
+    "native-engine/auron-planner",
     "native-engine/auron-memmgr",
 ]
 
@@ -48,7 +48,7 @@ overflow-checks = false
 [workspace.dependencies]
 auron = { path = "./native-engine/auron" }
 auron-jni-bridge = { path = "./native-engine/auron-jni-bridge" }
-auron-serde = { path = "./native-engine/auron-serde" }
+auron-planner = { path = "./native-engine/auron-planner" }
 auron-memmgr = { path = "./native-engine/auron-memmgr" }
 datafusion-ext-commons = { path = "./native-engine/datafusion-ext-commons" }
 datafusion-ext-exprs = { path = "./native-engine/datafusion-ext-exprs" }

dev/mvn-build-helper/proto/pom.xml

@@ -48,7 +48,7 @@
         <version>0.6.1</version>
         <extensions>true</extensions>
         <configuration>
-          <protoSourceRoot>../../../native-engine/auron-serde/proto</protoSourceRoot>
+          <protoSourceRoot>../../../native-engine/auron-planner/proto</protoSourceRoot>
           <protocArtifact>com.google.protobuf:protoc:${protobufVersion}:exe:${os.detected.classifier}</protocArtifact>
           <checkStaleness>true</checkStaleness>
           <clearOutputDirectory>true</clearOutputDirectory>

native-engine/auron-serde/Cargo.toml → native-engine/auron-planner/Cargo.toml

@@ -16,7 +16,7 @@
 #
 
 [package]
-name = "auron-serde"
+name = "auron-planner"
 version = "0.1.0"
 edition = "2024"
 

native-engine/auron-serde/src/lib.rs → native-engine/auron-planner/src/lib.rs

@@ -28,7 +28,7 @@ pub mod protobuf {
 }
 
 pub mod error;
-pub mod from_proto;
+pub mod planner;
 
 pub(crate) fn proto_error<S: Into<String>>(message: S) -> PlanSerDeError {
     PlanSerDeError::General(message.into())
@@ -58,9 +58,9 @@ macro_rules! into_required {
 
 #[macro_export]
 macro_rules! convert_box_required {
-    ($PB:expr) => {{
+    ($self:expr, $PB:expr) => {{
         if let Some(field) = $PB.as_ref() {
-            field.as_ref().try_into()
+            $self.create_plan(field.as_ref())
         } else {
             Err(proto_error("Missing required field in protobuf"))
         }

native-engine/auron/Cargo.toml

@@ -34,7 +34,7 @@ http-service = []
 arrow = { workspace = true }
 auron-jni-bridge = { workspace = true }
 auron-memmgr = { workspace = true }
-auron-serde = { workspace = true }
+auron-planner = { workspace = true }
 datafusion = { workspace = true }
 datafusion-ext-commons = { workspace = true }
 datafusion-ext-plans = { workspace = true }

native-engine/auron/src/rt.rs

@@ -29,7 +29,7 @@ use auron_jni_bridge::{
     is_task_running, jni_call, jni_call_static, jni_convert_byte_array, jni_exception_check,
     jni_exception_occurred, jni_new_global_ref, jni_new_object, jni_new_string,
 };
-use auron_serde::protobuf::TaskDefinition;
+use auron_planner::{planner::PhysicalPlanner, protobuf::TaskDefinition};
 use datafusion::{
     common::Result,
     error::DataFusionError,
@@ -83,9 +83,11 @@ impl NativeExecutionRuntime {
         let plan = &task_definition.plan.expect("plan is empty");
         drop(raw_task_definition);
 
+        let planner = PhysicalPlanner::new(partition_id);
+
         // get execution plan
-        let execution_plan: Arc<dyn ExecutionPlan> = plan
-            .try_into()
+        let execution_plan: Arc<dyn ExecutionPlan> = planner
+            .create_plan(plan)
             .or_else(|err| df_execution_err!("cannot create execution plan: {err:?}"))?;
 
         let exec_ctx = ExecutionContext::new(

native-engine/datafusion-ext-functions/src/lib.rs

@@ -34,7 +34,10 @@ mod spark_round;
 mod spark_strings;
 mod spark_unscaled_value;
 
-pub fn create_auron_ext_function(name: &str) -> Result<ScalarFunctionImplementation> {
+pub fn create_auron_ext_function(
+    name: &str,
+    spark_partition_id: usize,
+) -> Result<ScalarFunctionImplementation> {
     // auron ext functions, if used for spark should be start with 'Spark_',
     // if used for flink should be start with 'Flink_',
     // same to other engines.