chore: Simplifying BlockMetaInfo Definitions (#16716)

* clean up

Signed-off-by: coldWater <forsaken628@gmail.com>

* boxed

Signed-off-by: coldWater <forsaken628@gmail.com>

---------

Signed-off-by: coldWater <forsaken628@gmail.com>

Author: forsaken628

src/query/catalog/src/plan/agg_index.rs

@@ -14,6 +14,7 @@
 
 use std::fmt::Debug;
 
+use databend_common_expression::local_block_meta_serde;
 use databend_common_expression::BlockMetaInfo;
 use databend_common_expression::BlockMetaInfoPtr;
 use databend_common_expression::RemoteExpr;
@@ -69,26 +70,10 @@ impl AggIndexMeta {
     }
 }
 
-impl serde::Serialize for AggIndexMeta {
-    fn serialize<S>(&self, _: S) -> std::result::Result<S::Ok, S::Error>
-    where S: serde::Serializer {
-        unimplemented!("Unimplemented serialize AggIndexMeta")
-    }
-}
-
-impl<'de> serde::Deserialize<'de> for AggIndexMeta {
-    fn deserialize<D>(_: D) -> std::result::Result<Self, D::Error>
-    where D: serde::Deserializer<'de> {
-        unimplemented!("Unimplemented deserialize AggIndexMeta")
-    }
-}
+local_block_meta_serde!(AggIndexMeta);
 
 #[typetag::serde(name = "agg_index_meta")]
 impl BlockMetaInfo for AggIndexMeta {
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("Unimplemented equals AggIndexMeta")
-    }
-
     fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
         Box::new(self.clone())
     }

src/query/expression/src/block.rs

@@ -83,29 +83,37 @@ impl BlockEntry {
 #[typetag::serde(tag = "type")]
 pub trait BlockMetaInfo: Debug + Send + Sync + Any + 'static {
     #[allow(clippy::borrowed_box)]
-    fn equals(&self, info: &Box<dyn BlockMetaInfo>) -> bool;
+    fn equals(&self, _info: &Box<dyn BlockMetaInfo>) -> bool {
+        panic!(
+            "The reason for not implementing equals is usually because the higher-level logic doesn't allow/need the meta to be compared."
+        )
+    }
 
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo>;
+    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
+        panic!(
+            "The reason for not implementing clone_self is usually because the higher-level logic doesn't allow/need the associated block to be cloned."
+        )
+    }
 }
 
-pub trait BlockMetaInfoDowncast: Sized {
-    fn downcast_from(boxed: BlockMetaInfoPtr) -> Option<Self>;
-
-    fn downcast_ref_from(boxed: &BlockMetaInfoPtr) -> Option<&Self>;
-}
+pub trait BlockMetaInfoDowncast: Sized + BlockMetaInfo {
+    fn boxed(self) -> BlockMetaInfoPtr {
+        Box::new(self)
+    }
 
-impl<T: BlockMetaInfo> BlockMetaInfoDowncast for T {
     fn downcast_from(boxed: BlockMetaInfoPtr) -> Option<Self> {
         let boxed: Box<dyn Any> = boxed;
         boxed.downcast().ok().map(|x| *x)
     }
 
     fn downcast_ref_from(boxed: &BlockMetaInfoPtr) -> Option<&Self> {
-        let boxed: &dyn Any = boxed.as_ref();
+        let boxed = boxed.as_ref() as &dyn Any;
         boxed.downcast_ref()
     }
 }
 
+impl<T: BlockMetaInfo> BlockMetaInfoDowncast for T {}
+
 impl DataBlock {
     #[inline]
     pub fn new(columns: Vec<BlockEntry>, num_rows: usize) -> Self {
@@ -633,8 +641,8 @@ impl Eq for Box<dyn BlockMetaInfo> {}
 
 impl PartialEq for Box<dyn BlockMetaInfo> {
     fn eq(&self, other: &Self) -> bool {
-        let this_any: &dyn Any = self.as_ref();
-        let other_any: &dyn Any = other.as_ref();
+        let this_any = self.as_ref() as &dyn Any;
+        let other_any = other.as_ref() as &dyn Any;
 
         match this_any.type_id() == other_any.type_id() {
             true => self.equals(other),
@@ -668,3 +676,28 @@ fn check_type(data_type: &DataType, value: &Value<AnyType>) {
         Value::Column(c) => assert_eq!(&c.data_type(), data_type),
     }
 }
+
+#[macro_export]
+macro_rules! local_block_meta_serde {
+    ($T:ty) => {
+        impl serde::Serialize for $T {
+            fn serialize<S>(&self, _: S) -> std::result::Result<S::Ok, S::Error>
+            where S: serde::Serializer {
+                unreachable!(
+                    "{} must not be exchanged between multiple nodes.",
+                    stringify!($T)
+                )
+            }
+        }
+
+        impl<'de> serde::Deserialize<'de> for $T {
+            fn deserialize<D>(_: D) -> std::result::Result<Self, D::Error>
+            where D: serde::Deserializer<'de> {
+                unreachable!(
+                    "{} must not be exchanged between multiple nodes.",
+                    stringify!($T)
+                )
+            }
+        }
+    };
+}

src/query/pipeline/core/tests/it/pipelines/processors/port_test.rs

@@ -18,13 +18,12 @@ use std::sync::Barrier;
 use databend_common_base::runtime::Thread;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::local_block_meta_serde;
 use databend_common_expression::BlockMetaInfo;
 use databend_common_expression::DataBlock;
 use databend_common_pipeline_core::processors::connect;
 use databend_common_pipeline_core::processors::InputPort;
 use databend_common_pipeline_core::processors::OutputPort;
-use serde::Deserializer;
-use serde::Serializer;
 
 #[derive(Clone, Debug)]
 struct TestDataMeta {
@@ -43,26 +42,10 @@ impl TestDataMeta {
     }
 }
 
-impl serde::Serialize for TestDataMeta {
-    fn serialize<S>(&self, _: S) -> std::result::Result<S::Ok, S::Error>
-    where S: Serializer {
-        unimplemented!("Serialize is unimplemented for TestDataMeta")
-    }
-}
-
-impl<'de> serde::Deserialize<'de> for TestDataMeta {
-    fn deserialize<D>(_: D) -> std::result::Result<Self, D::Error>
-    where D: Deserializer<'de> {
-        unimplemented!("Deserialize is unimplemented for TestDataMeta")
-    }
-}
+local_block_meta_serde!(TestDataMeta);
 
 #[typetag::serde(name = "test_data_meta")]
 impl BlockMetaInfo for TestDataMeta {
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("equals is unimplemented for TestDataMeta")
-    }
-
     fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
         Box::new(self.clone())
     }

src/query/pipeline/transforms/src/processors/transforms/sort/k_way_merge_sort_partition.rs

@@ -18,8 +18,6 @@ use std::sync::Arc;
 
 use databend_common_exception::Result;
 use databend_common_expression::BlockMetaInfo;
-use databend_common_expression::BlockMetaInfoDowncast;
-use databend_common_expression::BlockMetaInfoPtr;
 use databend_common_expression::DataBlock;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::SortColumnDescription;
@@ -240,19 +238,5 @@ pub struct SortTaskMeta {
     pub input: usize,
 }
 
-impl SortTaskMeta {
-    pub fn as_meta(self) -> BlockMetaInfoPtr {
-        Box::new(self)
-    }
-}
-
 #[typetag::serde(name = "sort_task")]
-impl BlockMetaInfo for SortTaskMeta {
-    fn equals(&self, info: &Box<dyn BlockMetaInfo>) -> bool {
-        SortTaskMeta::downcast_ref_from(info).map_or(false, |info| self == info)
-    }
-
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
-        Box::new(*self)
-    }
-}
+impl BlockMetaInfo for SortTaskMeta {}

src/query/pipeline/transforms/src/processors/transforms/sort/spill.rs

@@ -12,36 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use databend_common_expression::local_block_meta_serde;
 use databend_common_expression::BlockMetaInfo;
 
 use super::super::SortSpillParams;
 
 /// Mark a partially sorted [`DataBlock`] as a block needs to be spilled.
-#[derive(serde::Serialize, serde::Deserialize, Debug)]
+#[derive(Debug)]
 pub struct SortSpillMetaWithParams(pub SortSpillParams);
 
-#[typetag::serde(name = "sort_spill")]
-impl BlockMetaInfo for SortSpillMetaWithParams {
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("Unimplemented equals SortSpillMetaWithParams")
-    }
+local_block_meta_serde!(SortSpillMetaWithParams);
 
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
-        unimplemented!("Unimplemented clone SortSpillMetaWithParams")
-    }
-}
+#[typetag::serde(name = "sort_spill")]
+impl BlockMetaInfo for SortSpillMetaWithParams {}
 
 /// Mark a partially sorted [`DataBlock`] as a block needs to be spilled.
-#[derive(serde::Serialize, serde::Deserialize, Debug)]
+#[derive(Debug)]
 pub struct SortSpillMeta {}
 
+local_block_meta_serde!(SortSpillMeta);
+
 #[typetag::serde(name = "sort_spill")]
-impl BlockMetaInfo for SortSpillMeta {
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("Unimplemented equals SortSpillMeta")
-    }
-
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
-        unimplemented!("Unimplemented clone SortSpillMeta")
-    }
-}
+impl BlockMetaInfo for SortSpillMeta {}

src/query/pipeline/transforms/src/processors/transforms/transform_compact_block.rs

@@ -21,6 +21,7 @@ use std::sync::Arc;
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::local_block_meta_serde;
 use databend_common_expression::BlockMetaInfo;
 use databend_common_expression::DataBlock;
 
@@ -42,30 +43,10 @@ impl Debug for BlockCompactMeta {
     }
 }
 
-impl serde::Serialize for BlockCompactMeta {
-    fn serialize<S>(&self, _: S) -> std::result::Result<S::Ok, S::Error>
-    where S: serde::Serializer {
-        unimplemented!("Unimplemented serialize BlockCompactMeta")
-    }
-}
-
-impl<'de> serde::Deserialize<'de> for BlockCompactMeta {
-    fn deserialize<D>(_: D) -> std::result::Result<Self, D::Error>
-    where D: serde::Deserializer<'de> {
-        unimplemented!("Unimplemented deserialize BlockCompactMeta")
-    }
-}
+local_block_meta_serde!(BlockCompactMeta);
 
 #[typetag::serde(name = "block_compact")]
-impl BlockMetaInfo for BlockCompactMeta {
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("Unimplemented equals BlockCompactMeta")
-    }
-
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
-        unimplemented!("Unimplemented clone BlockCompactMeta")
-    }
-}
+impl BlockMetaInfo for BlockCompactMeta {}
 
 #[derive(Default)]
 pub struct TransformCompactBlock {

src/query/pipeline/transforms/src/processors/transforms/transform_k_way_merge_sort.rs

@@ -532,7 +532,7 @@ where A: SortAlgorithm + 'static
                 total: task.total,
                 input: 0,
             }
-            .as_meta();
+            .boxed();
             self.output_data.push_back(block.add_meta(Some(meta))?);
         }
         debug_assert_eq!(rows, task.total);

src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_meta.rs

@@ -246,12 +246,4 @@ impl<Method: HashMethodBounds, V: Send + Sync + 'static> BlockMetaInfo
     fn typetag_name(&self) -> &'static str {
         unimplemented!("AggregateMeta does not support exchanging between multiple nodes")
     }
-
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("Unimplemented equals for AggregateMeta")
-    }
-
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
-        unimplemented!("Unimplemented clone for AggregateMeta")
-    }
 }

src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_exchange_group_by_serializer.rs

@@ -26,6 +26,7 @@ use databend_common_base::runtime::profile::ProfileStatisticsName;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
 use databend_common_expression::arrow::serialize_column;
+use databend_common_expression::local_block_meta_serde;
 use databend_common_expression::types::ArgType;
 use databend_common_expression::types::ArrayType;
 use databend_common_expression::types::Int64Type;
@@ -140,30 +141,10 @@ impl Debug for FlightSerializedMeta {
     }
 }
 
-impl serde::Serialize for FlightSerializedMeta {
-    fn serialize<S>(&self, _: S) -> std::result::Result<S::Ok, S::Error>
-    where S: serde::Serializer {
-        unimplemented!("Unimplemented serialize FlightSerializedMeta")
-    }
-}
-
-impl<'de> serde::Deserialize<'de> for FlightSerializedMeta {
-    fn deserialize<D>(_: D) -> std::result::Result<Self, D::Error>
-    where D: serde::Deserializer<'de> {
-        unimplemented!("Unimplemented deserialize FlightSerializedMeta")
-    }
-}
+local_block_meta_serde!(FlightSerializedMeta);
 
 #[typetag::serde(name = "exchange_shuffle")]
-impl BlockMetaInfo for FlightSerializedMeta {
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("Unimplemented equals FlightSerializedMeta")
-    }
-
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
-        unimplemented!("Unimplemented clone FlightSerializedMeta")
-    }
-}
+impl BlockMetaInfo for FlightSerializedMeta {}
 
 impl<Method: HashMethodBounds> BlockMetaTransform<ExchangeShuffleMeta>
     for TransformExchangeGroupBySerializer<Method>

src/query/service/src/pipelines/processors/transforms/window/partition/window_partition_meta.rs

@@ -15,6 +15,7 @@
 use std::fmt::Debug;
 use std::fmt::Formatter;
 
+use databend_common_expression::local_block_meta_serde;
 use databend_common_expression::BlockMetaInfo;
 use databend_common_expression::BlockMetaInfoPtr;
 use databend_common_expression::DataBlock;
@@ -30,40 +31,13 @@ impl WindowPartitionMeta {
     }
 }
 
-impl serde::Serialize for WindowPartitionMeta {
-    fn serialize<S>(&self, _: S) -> Result<S::Ok, S::Error>
-    where S: serde::Serializer {
-        unreachable!("WindowPartitionMeta does not support exchanging between multiple nodes")
-    }
-}
-
-impl<'de> serde::Deserialize<'de> for WindowPartitionMeta {
-    fn deserialize<D>(_: D) -> Result<Self, D::Error>
-    where D: serde::Deserializer<'de> {
-        unreachable!("WindowPartitionMeta does not support exchanging between multiple nodes")
-    }
-}
+local_block_meta_serde!(WindowPartitionMeta);
 
 impl Debug for WindowPartitionMeta {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("WindowPartitionMeta").finish()
     }
 }
 
-impl BlockMetaInfo for WindowPartitionMeta {
-    fn typetag_deserialize(&self) {
-        unimplemented!("WindowPartitionMeta does not support exchanging between multiple nodes")
-    }
-
-    fn typetag_name(&self) -> &'static str {
-        unimplemented!("WindowPartitionMeta does not support exchanging between multiple nodes")
-    }
-
-    fn equals(&self, _: &Box<dyn BlockMetaInfo>) -> bool {
-        unimplemented!("Unimplemented equals for WindowPartitionMeta")
-    }
-
-    fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
-        unimplemented!("Unimplemented clone for WindowPartitionMeta")
-    }
-}
+#[typetag::serde(name = "window_partition")]
+impl BlockMetaInfo for WindowPartitionMeta {}