update

Author: luoluoyuyu

src/runtime/streaming/operators/grouping/incremental_aggregate.rs

@@ -49,7 +49,7 @@ use crate::runtime::streaming::arrow::decode_aggregate;
 use crate::runtime::streaming::operators::{Key, UpdatingCache};
 use crate::runtime::streaming::StreamOutput;
 use crate::sql::common::{to_nanos, CheckpointBarrier, FsSchema, Watermark, TIMESTAMP_FIELD, UPDATING_META_FIELD};
-use crate::sql::logical_planner::updating_meta_fields;
+use crate::sql::physical::updating_meta_fields;
 
 #[derive(Debug, Copy, Clone)]
 struct BatchData {

src/runtime/streaming/operators/joins/join_instance.rs

@@ -34,8 +34,9 @@ use crate::runtime::streaming::api::operator::{MessageOperator, Registry};
 use async_trait::async_trait;
 use protocol::grpc::api::JoinOperator;
 use crate::runtime::streaming::StreamOutput;
+use crate::sql::common::constants::mem_exec_join_side;
 use crate::sql::common::{from_nanos, CheckpointBarrier, FsSchema, FsSchemaRef, Watermark};
-use crate::sql::logical_planner::{DecodingContext, FsPhysicalExtensionCodec};
+use crate::sql::physical::{DecodingContext, FsPhysicalExtensionCodec};
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 enum JoinSide {
@@ -47,8 +48,8 @@ impl JoinSide {
     #[allow(dead_code)]
     fn name(&self) -> &'static str {
         match self {
-            JoinSide::Left => "left",
-            JoinSide::Right => "right",
+            JoinSide::Left => mem_exec_join_side::LEFT,
+            JoinSide::Right => mem_exec_join_side::RIGHT,
         }
     }
 }

src/runtime/streaming/operators/joins/join_with_expiration.rs

@@ -33,7 +33,7 @@ use async_trait::async_trait;
 use protocol::grpc::api::JoinOperator;
 use crate::runtime::streaming::StreamOutput;
 use crate::sql::common::{CheckpointBarrier, FsSchema, Watermark};
-use crate::sql::logical_planner::{DecodingContext, FsPhysicalExtensionCodec};
+use crate::sql::physical::{DecodingContext, FsPhysicalExtensionCodec};
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 enum JoinSide {

src/runtime/streaming/operators/stateless_physical_executor.rs

@@ -27,7 +27,7 @@ use futures::StreamExt;
 use prost::Message;
 
 use crate::runtime::streaming::api::operator::Registry;
-use crate::sql::logical_planner::{DecodingContext, FsPhysicalExtensionCodec};
+use crate::sql::physical::{DecodingContext, FsPhysicalExtensionCodec};
 
 pub struct StatelessPhysicalExecutor {
     batch: Arc<RwLock<Option<RecordBatch>>>,

src/runtime/streaming/operators/windows/session_aggregating_window.rs

@@ -44,7 +44,7 @@ use protocol::grpc::api::SessionWindowAggregateOperator;
 use crate::runtime::streaming::StreamOutput;
 use crate::sql::common::{from_nanos, to_nanos, CheckpointBarrier, FsSchema, FsSchemaRef, Watermark};
 use crate::sql::common::converter::Converter;
-use crate::sql::logical_planner::{DecodingContext, FsPhysicalExtensionCodec};
+use crate::sql::physical::{DecodingContext, FsPhysicalExtensionCodec};
 use crate::sql::schema::utils::window_arrow_struct;
 // ============================================================================
 // 领域模型与纯内存状态

src/runtime/streaming/operators/windows/sliding_aggregating_window.rs

@@ -43,7 +43,7 @@ use crate::runtime::streaming::api::operator::Registry;
 use protocol::grpc::api::SlidingWindowAggregateOperator;
 use crate::runtime::streaming::StreamOutput;
 use crate::sql::common::{from_nanos, to_nanos, CheckpointBarrier, FsSchema, Watermark};
-use crate::sql::logical_planner::{DecodingContext, FsPhysicalExtensionCodec};
+use crate::sql::physical::{DecodingContext, FsPhysicalExtensionCodec};
 // ============================================================================
 // 纯内存状态：阶梯式时间面板 (Tiered panes)
 // 这部分本身就是极佳的内存数据结构，原样保留！

src/runtime/streaming/operators/windows/tumbling_aggregating_window.rs

@@ -44,7 +44,7 @@ use protocol::grpc::api::TumblingWindowAggregateOperator;
 use crate::runtime::streaming::StreamOutput;
 use crate::sql::common::{from_nanos, to_nanos, CheckpointBarrier, FsSchema, Watermark};
 use crate::sql::common::time_utils::print_time;
-use crate::sql::logical_planner::{DecodingContext, FsPhysicalExtensionCodec};
+use crate::sql::physical::{DecodingContext, FsPhysicalExtensionCodec};
 use crate::sql::schema::utils::add_timestamp_field_arrow;
 
 struct ActiveBin {

src/runtime/streaming/operators/windows/window_function.rs

@@ -37,7 +37,7 @@ use async_trait::async_trait;
 use crate::runtime::streaming::StreamOutput;
 use crate::sql::common::{from_nanos, CheckpointBarrier, FsSchema, FsSchemaRef, Watermark};
 use crate::sql::common::time_utils::print_time;
-use crate::sql::logical_planner::{DecodingContext, FsPhysicalExtensionCodec};
+use crate::sql::physical::{DecodingContext, FsPhysicalExtensionCodec};
 
 // ============================================================================
 // 纯内存执行缓冲区

src/sql/analysis/async_udf_rewriter.rs

@@ -11,7 +11,8 @@
 // limitations under the License.
 
 use crate::sql::extensions::remote_table::RemoteTableBoundaryNode;
-use crate::sql::extensions::{ASYNC_RESULT_FIELD, AsyncFunctionExecutionNode};
+use crate::sql::common::constants::sql_field;
+use crate::sql::extensions::AsyncFunctionExecutionNode;
 use crate::sql::schema::StreamSchemaProvider;
 use datafusion::common::tree_node::{Transformed, TreeNode, TreeNodeRewriter};
 use datafusion::common::{Column, Result as DFResult, TableReference, plan_err};
@@ -55,7 +56,7 @@ impl<'a> AsyncUdfRewriter<'a> {
                         );
                     }
                     return Ok(Transformed::yes(Expr::Column(Column::new_unqualified(
-                        ASYNC_RESULT_FIELD,
+                        sql_field::ASYNC_RESULT,
                     ))));
                 }
             }

src/sql/analysis/join_rewriter.rs

@@ -15,6 +15,7 @@ use crate::sql::extensions::join::StreamingJoinNode;
 use crate::sql::extensions::key_calculation::KeyExtractionNode;
 use crate::sql::analysis::streaming_window_analzer::StreamingWindowAnalzer;
 use crate::sql::types::{WindowType, fields_with_qualifiers, schema_from_df_fields_with_metadata};
+use crate::sql::common::constants::mem_exec_join_side;
 use crate::sql::common::TIMESTAMP_FIELD;
 use datafusion::common::tree_node::{Transformed, TreeNodeRewriter};
 use datafusion::common::{
@@ -198,8 +199,8 @@ impl TreeNodeRewriter for JoinRewriter<'_> {
 
         // 2. Prepare Keyed Inputs for Shuffle
         let (left_on, right_on): (Vec<_>, Vec<_>) = join.on.clone().into_iter().unzip();
-        let keyed_left = self.build_keyed_side(join.left, left_on, "left")?;
-        let keyed_right = self.build_keyed_side(join.right, right_on, "right")?;
+        let keyed_left = self.build_keyed_side(join.left, left_on, mem_exec_join_side::LEFT)?;
+        let keyed_right = self.build_keyed_side(join.right, right_on, mem_exec_join_side::RIGHT)?;
 
         // 3. Assemble Rewritten Join Node
         let join_schema = Arc::new(build_join_schema(