Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/proto/batch_plan.proto b/proto/batch_plan.proto
index 77cdcb08f7e2..2373b7d483e3 100644
--- a/proto/batch_plan.proto
+++ b/proto/batch_plan.proto
@@ -48,7 +48,7 @@ message ScanRange {
 
   // `None` represent unbounded.
   message Bound {
-    bytes value = 1;
+    repeated bytes value = 1;
     bool inclusive = 2;
   }
   //  The lower bound of the next PK column subsequent to those in `eq_conds`.
diff --git a/src/batch/src/executor/row_seq_scan.rs b/src/batch/src/executor/row_seq_scan.rs
index b65f4bf8939b..6f65d3fc16c7 100644
--- a/src/batch/src/executor/row_seq_scan.rs
+++ b/src/batch/src/executor/row_seq_scan.rs
@@ -23,7 +23,7 @@ use risingwave_common::bitmap::Bitmap;
 use risingwave_common::catalog::{ColumnId, Schema};
 use risingwave_common::hash::VnodeCountCompat;
 use risingwave_common::row::{OwnedRow, Row};
-use risingwave_common::types::{DataType, Datum};
+use risingwave_common::types::DataType;
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::value_encoding::deserialize_datum;
 use risingwave_pb::batch_plan::plan_node::NodeBody;
@@ -60,12 +60,13 @@ pub struct RowSeqScanExecutor<S: StateStore> {
 }
 
 /// Range for batch scan.
+#[derive(Debug)]
 pub struct ScanRange {
     /// The prefix of the primary key.
     pub pk_prefix: OwnedRow,
 
     /// The range bounds of the next column.
-    pub next_col_bounds: (Bound<Datum>, Bound<Datum>),
+    pub next_col_bounds: (Bound<OwnedRow>, Bound<OwnedRow>),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -100,17 +101,16 @@ impl From<&AsOf> for PbAsOf {
 
 impl ScanRange {
     /// Create a scan range from the prost representation.
-    pub fn new(
-        scan_range: PbScanRange,
-        mut pk_types: impl Iterator<Item = DataType>,
-    ) -> Result<Self> {
+    pub fn new(scan_range: PbScanRange, pk_types: Vec<DataType>) -> Result<Self> {
+        let mut index = 0;
         let pk_prefix = OwnedRow::new(
             scan_range
                 .eq_conds
                 .iter()
                 .map(|v| {
-                    let ty = pk_types.next().unwrap();
-                    deserialize_datum(v.as_slice(), &ty)
+                    let ty = pk_types.get(index).unwrap();
+                    index += 1;
+                    deserialize_datum(v.as_slice(), ty)
                 })
                 .try_collect()?,
         );
@@ -121,26 +121,34 @@ impl ScanRange {
             });
         }
 
-        let bound_ty = pk_types.next().unwrap();
-        let build_bound = |bound: &scan_range::Bound| -> Bound<Datum> {
-            let datum = deserialize_datum(bound.value.as_slice(), &bound_ty).unwrap();
+        let build_bound = |bound: &scan_range::Bound, mut index| -> Result<Bound<OwnedRow>> {
+            let next_col_bounds = OwnedRow::new(
+                bound
+                    .value
+                    .iter()
+                    .map(|v| {
+                        let ty = pk_types.get(index).unwrap();
+                        index += 1;
+                        deserialize_datum(v.as_slice(), ty)
+                    })
+                    .try_collect()?,
+            );
             if bound.inclusive {
-                Bound::Included(datum)
+                Ok(Bound::Included(next_col_bounds))
             } else {
-                Bound::Excluded(datum)
+                Ok(Bound::Excluded(next_col_bounds))
             }
         };
 
-        let next_col_bounds: (Bound<Datum>, Bound<Datum>) = match (
+        let next_col_bounds: (Bound<OwnedRow>, Bound<OwnedRow>) = match (
             scan_range.lower_bound.as_ref(),
             scan_range.upper_bound.as_ref(),
         ) {
-            (Some(lb), Some(ub)) => (build_bound(lb), build_bound(ub)),
-            (None, Some(ub)) => (Bound::Unbounded, build_bound(ub)),
-            (Some(lb), None) => (build_bound(lb), Bound::Unbounded),
+            (Some(lb), Some(ub)) => (build_bound(lb, index)?, build_bound(ub, index)?),
+            (None, Some(ub)) => (Bound::Unbounded, build_bound(ub, index)?),
+            (Some(lb), None) => (build_bound(lb, index)?, Bound::Unbounded),
             (None, None) => unreachable!(),
         };
-
         Ok(Self {
             pk_prefix,
             next_col_bounds,
@@ -221,14 +229,18 @@ impl BoxedExecutorBuilder for RowSeqScanExecutorBuilder {
                 scan_ranges
                     .iter()
                     .map(|scan_range| {
-                        let pk_types = table_desc.pk.iter().map(|order| {
-                            DataType::from(
-                                table_desc.columns[order.column_index as usize]
-                                    .column_type
-                                    .as_ref()
-                                    .unwrap(),
-                            )
-                        });
+                        let pk_types = table_desc
+                            .pk
+                            .iter()
+                            .map(|order| {
+                                DataType::from(
+                                    table_desc.columns[order.column_index as usize]
+                                        .column_type
+                                        .as_ref()
+                                        .unwrap(),
+                                )
+                            })
+                            .collect_vec();
                         ScanRange::new(scan_range.clone(), pk_types)
                     })
                     .try_collect()?
@@ -434,40 +446,39 @@ impl<S: StateStore> RowSeqScanExecutor<S> {
         let start_bound_is_bounded = !matches!(start_bound, Bound::Unbounded);
         let end_bound_is_bounded = !matches!(end_bound, Bound::Unbounded);
 
+        let build_bound = |other_bound_is_bounded: bool, bound, order_type_nulls| {
+            match bound {
+                Bound::Unbounded => {
+                    if other_bound_is_bounded && order_type_nulls {
+                        // `NULL`s are at the start bound side, we should exclude them to meet SQL semantics.
+                        Bound::Excluded(OwnedRow::new(vec![None]))
+                    } else {
+                        // Both start and end are unbounded, so we need to select all rows.
+                        Bound::Unbounded
+                    }
+                }
+                Bound::Included(x) => Bound::Included(x),
+                Bound::Excluded(x) => Bound::Excluded(x),
+            }
+        };
+        let start_bound = build_bound(
+            end_bound_is_bounded,
+            start_bound,
+            order_type.nulls_are_first(),
+        );
+        let end_bound = build_bound(
+            start_bound_is_bounded,
+            end_bound,
+            order_type.nulls_are_last(),
+        );
+
         // Range Scan.
         assert!(pk_prefix.len() < table.pk_indices().len());
         let iter = table
             .batch_chunk_iter_with_pk_bounds(
                 epoch.into(),
                 &pk_prefix,
-                (
-                    match start_bound {
-                        Bound::Unbounded => {
-                            if end_bound_is_bounded && order_type.nulls_are_first() {
-                                // `NULL`s are at the start bound side, we should exclude them to meet SQL semantics.
-                                Bound::Excluded(OwnedRow::new(vec![None]))
-                            } else {
-                                // Both start and end are unbounded, so we need to select all rows.
-                                Bound::Unbounded
-                            }
-                        }
-                        Bound::Included(x) => Bound::Included(OwnedRow::new(vec![x])),
-                        Bound::Excluded(x) => Bound::Excluded(OwnedRow::new(vec![x])),
-                    },
-                    match end_bound {
-                        Bound::Unbounded => {
-                            if start_bound_is_bounded && order_type.nulls_are_last() {
-                                // `NULL`s are at the end bound side, we should exclude them to meet SQL semantics.
-                                Bound::Excluded(OwnedRow::new(vec![None]))
-                            } else {
-                                // Both start and end are unbounded, so we need to select all rows.
-                                Bound::Unbounded
-                            }
-                        }
-                        Bound::Included(x) => Bound::Included(OwnedRow::new(vec![x])),
-                        Bound::Excluded(x) => Bound::Excluded(OwnedRow::new(vec![x])),
-                    },
-                ),
+                (start_bound, end_bound),
                 ordered,
                 chunk_size,
                 PrefetchOptions::new(limit.is_none(), true),
diff --git a/src/common/src/util/scan_range.rs b/src/common/src/util/scan_range.rs
index cfe209cf2c22..b45db071be3c 100644
--- a/src/common/src/util/scan_range.rs
+++ b/src/common/src/util/scan_range.rs
@@ -18,27 +18,39 @@ use paste::paste;
 use risingwave_pb::batch_plan::scan_range::Bound as PbBound;
 use risingwave_pb::batch_plan::ScanRange as PbScanRange;
 
-use super::value_encoding::serialize_datum;
 use crate::hash::table_distribution::TableDistribution;
 use crate::hash::VirtualNode;
 use crate::types::{Datum, ScalarImpl};
 use crate::util::value_encoding::serialize_datum_into;
 
-/// See also [`PbScanRange`]
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct ScanRange {
     pub eq_conds: Vec<Datum>,
-    pub range: (Bound<ScalarImpl>, Bound<ScalarImpl>),
+    pub range: (Bound<Vec<Datum>>, Bound<Vec<Datum>>),
 }
 
-fn bound_to_proto(bound: &Bound<ScalarImpl>) -> Option<PbBound> {
+fn bound_vec_datum_to_proto(bound: &Bound<Vec<Datum>>) -> Option<PbBound> {
     match bound {
         Bound::Included(literal) => Some(PbBound {
-            value: serialize_datum(Some(literal)),
+            value: literal
+                .iter()
+                .map(|datum| {
+                    let mut encoded = vec![];
+                    serialize_datum_into(datum, &mut encoded);
+                    encoded
+                })
+                .collect(),
             inclusive: true,
         }),
         Bound::Excluded(literal) => Some(PbBound {
-            value: serialize_datum(Some(literal)),
+            value: literal
+                .iter()
+                .map(|datum| {
+                    let mut encoded = vec![];
+                    serialize_datum_into(datum, &mut encoded);
+                    encoded
+                })
+                .collect(),
             inclusive: false,
         }),
         Bound::Unbounded => None,
@@ -57,8 +69,8 @@ impl ScanRange {
                     encoded
                 })
                 .collect(),
-            lower_bound: bound_to_proto(&self.range.0),
-            upper_bound: bound_to_proto(&self.range.1),
+            lower_bound: bound_vec_datum_to_proto(&self.range.0),
+            upper_bound: bound_vec_datum_to_proto(&self.range.1),
         }
     }
 
@@ -76,16 +88,16 @@ impl ScanRange {
             && !matches!(bounds.end_bound(), Bound::Unbounded)
     }
 
+    pub fn try_compute_vnode(&self, table_distribution: &TableDistribution) -> Option<VirtualNode> {
+        table_distribution.try_compute_vnode_by_pk_prefix(self.eq_conds.as_slice())
+    }
+
     pub const fn full_table_scan() -> Self {
         Self {
             eq_conds: vec![],
             range: full_range(),
         }
     }
-
-    pub fn try_compute_vnode(&self, table_distribution: &TableDistribution) -> Option<VirtualNode> {
-        table_distribution.try_compute_vnode_by_pk_prefix(self.eq_conds.as_slice())
-    }
 }
 
 pub const fn full_range<T>() -> (Bound<T>, Bound<T>) {
@@ -114,27 +126,30 @@ macro_rules! impl_split_small_range {
                 /// `Precondition`: make sure the first order key is int type if you call this method.
                 /// Optimize small range scan. It turns x between 0 and 5 into x in (0, 1, 2, 3, 4, 5).s
                 pub fn split_small_range(&self, max_gap: u64) -> Option<Vec<Self>> {
-                     if self.eq_conds.is_empty() {
-                        match self.range {
-                            $(
-                                (
-                                    Bound::Included(ScalarImpl::$type_name(ref left)),
-                                    Bound::Included(ScalarImpl::$type_name(ref right)),
-                                ) => {
-                                    if (right - left + 1) as u64 <= max_gap {
-                                        return Some(
-                                            (*left..=*right)
-                                                .into_iter()
-                                                .map(|i| ScanRange {
-                                                    eq_conds: vec![Some(ScalarImpl::$type_name(i))],
-                                                    range: full_range(),
-                                                })
-                                                .collect(),
-                                        );
+                    if self.eq_conds.is_empty() {
+                        if let (Bound::Included(left),Bound::Included(right)) = (&self.range.0, &self.range.1){
+                            match (left.get(0),right.get(0)) {
+                                $(
+                                    (
+                                        Some(Some(ScalarImpl::$type_name(ref left))),
+                                        Some(Some(ScalarImpl::$type_name(ref right))),
+                                    ) => {
+
+                                        if (right - left + 1) as u64 <= max_gap {
+                                            return Some(
+                                                (*left..=*right)
+                                                    .into_iter()
+                                                    .map(|i| ScanRange {
+                                                        eq_conds: vec![Some(ScalarImpl::$type_name(i))],
+                                                        range: full_range(),
+                                                    })
+                                                    .collect(),
+                                            );
+                                        }
                                     }
-                                }
-                            )*
-                            _ => {}
+                                )*
+                                _ => {}
+                            }
                         }
                     }
 
@@ -151,7 +166,6 @@ for_all_scalar_int_variants! { impl_split_small_range }
 mod tests {
     use super::*;
     use crate::row::OwnedRow;
-
     // dist_key is prefix of pk
     #[test]
     fn test_vnode_prefix() {
diff --git a/src/frontend/src/expr/literal.rs b/src/frontend/src/expr/literal.rs
index 02bcb6544cf5..8a9881416693 100644
--- a/src/frontend/src/expr/literal.rs
+++ b/src/frontend/src/expr/literal.rs
@@ -93,6 +93,10 @@ impl Literal {
         &self.data
     }
 
+    pub fn get_data_type(&self) -> &Option<DataType> {
+        &self.data_type
+    }
+
     pub fn is_untyped(&self) -> bool {
         self.data_type.is_none()
     }
diff --git a/src/frontend/src/optimizer/plan_node/batch_seq_scan.rs b/src/frontend/src/optimizer/plan_node/batch_seq_scan.rs
index 576793f4dd45..addd83087322 100644
--- a/src/frontend/src/optimizer/plan_node/batch_seq_scan.rs
+++ b/src/frontend/src/optimizer/plan_node/batch_seq_scan.rs
@@ -12,18 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::ops::Bound;
-
-use itertools::Itertools;
 use pretty_xmlish::{Pretty, XmlNode};
-use risingwave_common::types::ScalarImpl;
 use risingwave_common::util::scan_range::{is_full_range, ScanRange};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 use risingwave_pb::batch_plan::RowSeqScanNode;
 use risingwave_sqlparser::ast::AsOf;
 
 use super::batch::prelude::*;
-use super::utils::{childless_record, to_pb_time_travel_as_of, Distill};
+use super::utils::{childless_record, scan_ranges_as_strs, to_pb_time_travel_as_of, Distill};
 use super::{generic, ExprRewritable, PlanBase, PlanRef, ToDistributedBatch};
 use crate::catalog::ColumnId;
 use crate::error::Result;
@@ -135,37 +131,6 @@ impl BatchSeqScan {
         &self.scan_ranges
     }
 
-    fn scan_ranges_as_strs(&self, verbose: bool) -> Vec<String> {
-        let order_names = match verbose {
-            true => self.core.order_names_with_table_prefix(),
-            false => self.core.order_names(),
-        };
-        let mut range_strs = vec![];
-
-        let explain_max_range = 20;
-        for scan_range in self.scan_ranges.iter().take(explain_max_range) {
-            #[expect(clippy::disallowed_methods)]
-            let mut range_str = scan_range
-                .eq_conds
-                .iter()
-                .zip(order_names.iter())
-                .map(|(v, name)| match v {
-                    Some(v) => format!("{} = {:?}", name, v),
-                    None => format!("{} IS NULL", name),
-                })
-                .collect_vec();
-            if !is_full_range(&scan_range.range) {
-                let i = scan_range.eq_conds.len();
-                range_str.push(range_to_string(&order_names[i], &scan_range.range))
-            }
-            range_strs.push(range_str.join(" AND "));
-        }
-        if self.scan_ranges.len() > explain_max_range {
-            range_strs.push("...".to_string());
-        }
-        range_strs
-    }
-
     pub fn limit(&self) -> &Option<u64> {
         &self.limit
     }
@@ -173,33 +138,6 @@ impl BatchSeqScan {
 
 impl_plan_tree_node_for_leaf! { BatchSeqScan }
 
-fn lb_to_string(name: &str, lb: &Bound<ScalarImpl>) -> String {
-    let (op, v) = match lb {
-        Bound::Included(v) => (">=", v),
-        Bound::Excluded(v) => (">", v),
-        Bound::Unbounded => unreachable!(),
-    };
-    format!("{} {} {:?}", name, op, v)
-}
-fn ub_to_string(name: &str, ub: &Bound<ScalarImpl>) -> String {
-    let (op, v) = match ub {
-        Bound::Included(v) => ("<=", v),
-        Bound::Excluded(v) => ("<", v),
-        Bound::Unbounded => unreachable!(),
-    };
-    format!("{} {} {:?}", name, op, v)
-}
-fn range_to_string(name: &str, range: &(Bound<ScalarImpl>, Bound<ScalarImpl>)) -> String {
-    match (&range.0, &range.1) {
-        (Bound::Unbounded, Bound::Unbounded) => unreachable!(),
-        (Bound::Unbounded, ub) => ub_to_string(name, ub),
-        (lb, Bound::Unbounded) => lb_to_string(name, lb),
-        (lb, ub) => {
-            format!("{} AND {}", lb_to_string(name, lb), ub_to_string(name, ub))
-        }
-    }
-}
-
 impl Distill for BatchSeqScan {
     fn distill<'a>(&self) -> XmlNode<'a> {
         let verbose = self.base.ctx().is_explain_verbose();
@@ -208,7 +146,11 @@ impl Distill for BatchSeqScan {
         vec.push(("columns", self.core.columns_pretty(verbose)));
 
         if !self.scan_ranges.is_empty() {
-            let range_strs = self.scan_ranges_as_strs(verbose);
+            let order_names = match verbose {
+                true => self.core.order_names_with_table_prefix(),
+                false => self.core.order_names(),
+            };
+            let range_strs = scan_ranges_as_strs(order_names, &self.scan_ranges);
             vec.push((
                 "scan_ranges",
                 Pretty::Array(range_strs.into_iter().map(Pretty::from).collect()),
diff --git a/src/frontend/src/optimizer/plan_node/batch_sys_seq_scan.rs b/src/frontend/src/optimizer/plan_node/batch_sys_seq_scan.rs
index 6068c1131626..a7133e47a218 100644
--- a/src/frontend/src/optimizer/plan_node/batch_sys_seq_scan.rs
+++ b/src/frontend/src/optimizer/plan_node/batch_sys_seq_scan.rs
@@ -12,18 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::ops::Bound;
-
-use itertools::Itertools;
 use pretty_xmlish::{Pretty, XmlNode};
-use risingwave_common::types::ScalarImpl;
 use risingwave_common::util::scan_range::{is_full_range, ScanRange};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 use risingwave_pb::batch_plan::SysRowSeqScanNode;
 use risingwave_pb::plan_common::PbColumnDesc;
 
 use super::batch::prelude::*;
-use super::utils::{childless_record, Distill};
+use super::utils::{childless_record, scan_ranges_as_strs, Distill};
 use super::{generic, ExprRewritable, PlanBase, PlanRef, ToBatchPb, ToDistributedBatch};
 use crate::error::Result;
 use crate::expr::{ExprRewriter, ExprVisitor};
@@ -91,68 +87,10 @@ impl BatchSysSeqScan {
     pub fn scan_ranges(&self) -> &[ScanRange] {
         &self.scan_ranges
     }
-
-    fn scan_ranges_as_strs(&self, verbose: bool) -> Vec<String> {
-        let order_names = match verbose {
-            true => self.core.order_names_with_table_prefix(),
-            false => self.core.order_names(),
-        };
-        let mut range_strs = vec![];
-
-        let explain_max_range = 20;
-        for scan_range in self.scan_ranges.iter().take(explain_max_range) {
-            #[expect(clippy::disallowed_methods)]
-            let mut range_str = scan_range
-                .eq_conds
-                .iter()
-                .zip(order_names.iter())
-                .map(|(v, name)| match v {
-                    Some(v) => format!("{} = {:?}", name, v),
-                    None => format!("{} IS NULL", name),
-                })
-                .collect_vec();
-            if !is_full_range(&scan_range.range) {
-                let i = scan_range.eq_conds.len();
-                range_str.push(range_to_string(&order_names[i], &scan_range.range))
-            }
-            range_strs.push(range_str.join(" AND "));
-        }
-        if self.scan_ranges.len() > explain_max_range {
-            range_strs.push("...".to_string());
-        }
-        range_strs
-    }
 }
 
 impl_plan_tree_node_for_leaf! { BatchSysSeqScan }
 
-fn lb_to_string(name: &str, lb: &Bound<ScalarImpl>) -> String {
-    let (op, v) = match lb {
-        Bound::Included(v) => (">=", v),
-        Bound::Excluded(v) => (">", v),
-        Bound::Unbounded => unreachable!(),
-    };
-    format!("{} {} {:?}", name, op, v)
-}
-fn ub_to_string(name: &str, ub: &Bound<ScalarImpl>) -> String {
-    let (op, v) = match ub {
-        Bound::Included(v) => ("<=", v),
-        Bound::Excluded(v) => ("<", v),
-        Bound::Unbounded => unreachable!(),
-    };
-    format!("{} {} {:?}", name, op, v)
-}
-fn range_to_string(name: &str, range: &(Bound<ScalarImpl>, Bound<ScalarImpl>)) -> String {
-    match (&range.0, &range.1) {
-        (Bound::Unbounded, Bound::Unbounded) => unreachable!(),
-        (Bound::Unbounded, ub) => ub_to_string(name, ub),
-        (lb, Bound::Unbounded) => lb_to_string(name, lb),
-        (lb, ub) => {
-            format!("{} AND {}", lb_to_string(name, lb), ub_to_string(name, ub))
-        }
-    }
-}
-
 impl Distill for BatchSysSeqScan {
     fn distill<'a>(&self) -> XmlNode<'a> {
         let verbose = self.base.ctx().is_explain_verbose();
@@ -161,7 +99,11 @@ impl Distill for BatchSysSeqScan {
         vec.push(("columns", self.core.columns_pretty(verbose)));
 
         if !self.scan_ranges.is_empty() {
-            let range_strs = self.scan_ranges_as_strs(verbose);
+            let order_names = match verbose {
+                true => self.core.order_names_with_table_prefix(),
+                false => self.core.order_names(),
+            };
+            let range_strs = scan_ranges_as_strs(order_names, &self.scan_ranges);
             vec.push((
                 "scan_ranges",
                 Pretty::Array(range_strs.into_iter().map(Pretty::from).collect()),
diff --git a/src/frontend/src/optimizer/plan_node/utils.rs b/src/frontend/src/optimizer/plan_node/utils.rs
index 4dfa4b46dafa..84b150c72689 100644
--- a/src/frontend/src/optimizer/plan_node/utils.rs
+++ b/src/frontend/src/optimizer/plan_node/utils.rs
@@ -14,6 +14,7 @@
 
 use std::collections::HashMap;
 use std::default::Default;
+use std::ops::Bound;
 use std::vec;
 
 use anyhow::anyhow;
@@ -28,6 +29,8 @@ use risingwave_common::constants::log_store::v2::{
     KV_LOG_STORE_PREDEFINED_COLUMNS, PK_ORDERING, VNODE_COLUMN_INDEX,
 };
 use risingwave_common::hash::VnodeCount;
+use risingwave_common::types::ScalarImpl;
+use risingwave_common::util::scan_range::{is_full_range, ScanRange};
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 
 use crate::catalog::table_catalog::TableType;
@@ -465,3 +468,93 @@ pub fn to_pb_time_travel_as_of(a: &Option<AsOf>) -> Result<Option<PbAsOf>> {
         as_of_type: Some(as_of_type),
     }))
 }
+
+pub fn scan_ranges_as_strs(order_names: Vec<String>, scan_ranges: &Vec<ScanRange>) -> Vec<String> {
+    let mut range_strs = vec![];
+
+    let explain_max_range = 20;
+    for scan_range in scan_ranges.iter().take(explain_max_range) {
+        #[expect(clippy::disallowed_methods)]
+        let mut range_str = scan_range
+            .eq_conds
+            .iter()
+            .zip(order_names.iter())
+            .map(|(v, name)| match v {
+                Some(v) => format!("{} = {:?}", name, v),
+                None => format!("{} IS NULL", name),
+            })
+            .collect_vec();
+
+        let len = scan_range.eq_conds.len();
+        if !is_full_range(&scan_range.range) {
+            let bound_range_str = match (&scan_range.range.0, &scan_range.range.1) {
+                (Bound::Unbounded, Bound::Unbounded) => unreachable!(),
+                (Bound::Unbounded, ub) => ub_to_string(&order_names[len..], ub),
+                (lb, Bound::Unbounded) => lb_to_string(&order_names[len..], lb),
+                (lb, ub) => format!(
+                    "{} AND {}",
+                    lb_to_string(&order_names[len..], lb),
+                    ub_to_string(&order_names[len..], ub)
+                ),
+            };
+            range_str.push(bound_range_str);
+        }
+        range_strs.push(range_str.join(" AND "));
+    }
+    if scan_ranges.len() > explain_max_range {
+        range_strs.push("...".to_string());
+    }
+    range_strs
+}
+
+pub fn ub_to_string(order_names: &[String], ub: &Bound<Vec<Option<ScalarImpl>>>) -> String {
+    match ub {
+        Bound::Included(v) => {
+            let (name, value) = row_to_string(order_names, v);
+            format!("{} <= {}", name, value)
+        }
+        Bound::Excluded(v) => {
+            let (name, value) = row_to_string(order_names, v);
+            format!("{} < {}", name, value)
+        }
+        Bound::Unbounded => unreachable!(),
+    }
+}
+
+pub fn lb_to_string(order_names: &[String], lb: &Bound<Vec<Option<ScalarImpl>>>) -> String {
+    match lb {
+        Bound::Included(v) => {
+            let (name, value) = row_to_string(order_names, v);
+            format!("{} >= {}", name, value)
+        }
+        Bound::Excluded(v) => {
+            let (name, value) = row_to_string(order_names, v);
+            format!("{} > {}", name, value)
+        }
+        Bound::Unbounded => unreachable!(),
+    }
+}
+
+pub fn row_to_string(
+    order_names: &[String],
+    struct_values: &Vec<Option<ScalarImpl>>,
+) -> (String, String) {
+    let mut names = vec![];
+    let mut values = vec![];
+    #[expect(clippy::disallowed_methods)]
+    for (name, value) in order_names.iter().zip(struct_values.iter()) {
+        names.push(name);
+        match value {
+            Some(v) => values.push(format!("{:?}", v)),
+            None => values.push("null".to_string()),
+        }
+    }
+    if names.len() == 1 {
+        (names[0].clone(), values[0].clone())
+    } else {
+        (
+            format!("({})", names.iter().join(", ")),
+            format!("({})", values.iter().join(", ")),
+        )
+    }
+}
diff --git a/src/frontend/src/utils/condition.rs b/src/frontend/src/utils/condition.rs
index f3f9738e2cae..e497ba821864 100644
--- a/src/frontend/src/utils/condition.rs
+++ b/src/frontend/src/utils/condition.rs
@@ -22,6 +22,7 @@ use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use risingwave_common::catalog::{Schema, TableDesc};
 use risingwave_common::types::{DataType, DefaultOrd, ScalarImpl};
+use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_common::util::scan_range::{is_full_range, ScanRange};
 
 use crate::error::Result;
@@ -357,6 +358,105 @@ impl Condition {
         }
     }
 
+    fn split_row_cmp_to_scan_ranges(
+        &self,
+        table_desc: Rc<TableDesc>,
+    ) -> Result<Option<(Vec<ScanRange>, Self)>> {
+        let (mut row_conjunctions, row_conjunctions_without_struct): (Vec<_>, Vec<_>) =
+            self.conjunctions.clone().into_iter().partition(|expr| {
+                if let Some(f) = expr.as_function_call() {
+                    if let Some(left_input) = f.inputs().get(0)
+                        && let Some(left_input) = left_input.as_function_call()
+                        && matches!(left_input.func_type(), ExprType::Row)
+                        && left_input.inputs().iter().all(|x| x.is_input_ref())
+                        && let Some(right_input) = f.inputs().get(1)
+                        && right_input.is_literal()
+                    {
+                        true
+                    } else {
+                        false
+                    }
+                } else {
+                    false
+                }
+            });
+
+        // optimize for single row conjunctions. More optimisations may come later
+        // For example, (v1,v2,v3) > (1, 2, 3) means all data from (1, 2, 3).
+        // Suppose v1 v2 v3 are both pk, we can push (v1,v2,v3）> (1,2,3) down to scan
+        // Suppose v1 v2 are both pk, we can push (v1,v2）> (1,2) down to scan and add (v1,v2,v3) > (1,2,3) in filter, it is still possible to reduce the value of scan
+        if row_conjunctions.len() == 1 {
+            let row_conjunction = row_conjunctions.pop().unwrap();
+            let row_left_inputs = row_conjunction
+                .as_function_call()
+                .unwrap()
+                .inputs()
+                .get(0)
+                .unwrap()
+                .as_function_call()
+                .unwrap()
+                .inputs();
+            let row_right_literal = row_conjunction
+                .as_function_call()
+                .unwrap()
+                .inputs()
+                .get(1)
+                .unwrap()
+                .as_literal()
+                .unwrap();
+            if !matches!(row_right_literal.get_data(), Some(ScalarImpl::Struct(_))) {
+                return Ok(None);
+            }
+            let row_right_literal_data = row_right_literal.get_data().clone().unwrap();
+            let right_iter = row_right_literal_data.as_struct().fields();
+            let func_type = row_conjunction.as_function_call().unwrap().func_type();
+            if row_left_inputs.len() > 1
+                && (matches!(func_type, ExprType::LessThan)
+                    || matches!(func_type, ExprType::GreaterThan))
+            {
+                let mut pk_struct = vec![];
+                let mut all_added = true;
+                let mut iter = row_left_inputs.iter().zip_eq_fast(right_iter);
+                for i in 0..table_desc.order_column_indices().len() {
+                    if let Some((left_expr, right_expr)) = iter.next() {
+                        if left_expr.as_input_ref().unwrap().index != i {
+                            all_added = false;
+                            break;
+                        }
+                        pk_struct.push(right_expr.clone());
+                    }
+                }
+
+                if !pk_struct.is_empty() {
+                    let scan_range = ScanRange {
+                        eq_conds: vec![],
+                        range: match func_type {
+                            ExprType::GreaterThan => (Bound::Excluded(pk_struct), Bound::Unbounded),
+                            ExprType::LessThan => (Bound::Unbounded, Bound::Excluded(pk_struct)),
+                            _ => unreachable!(),
+                        },
+                    };
+                    if !all_added {
+                        return Ok(Some((
+                            vec![scan_range],
+                            Condition {
+                                conjunctions: self.conjunctions.clone(),
+                            },
+                        )));
+                    } else {
+                        return Ok(Some((
+                            vec![scan_range],
+                            Condition {
+                                conjunctions: row_conjunctions_without_struct,
+                            },
+                        )));
+                    }
+                }
+            }
+        }
+        Ok(None)
+    }
+
     /// See also [`ScanRange`](risingwave_pb::batch_plan::ScanRange).
     pub fn split_to_scan_ranges(
         self,
@@ -381,12 +481,17 @@ impl Condition {
                 }
             }
         }
+        if let Some((scan_ranges, other_condition)) =
+            self.split_row_cmp_to_scan_ranges(table_desc.clone())?
+        {
+            return Ok((scan_ranges, other_condition));
+        }
 
         let mut groups = Self::classify_conjunctions_by_pk(self.conjunctions, &table_desc);
+        let mut other_conds = groups.pop().unwrap();
 
         // Analyze each group and use result to update scan range.
         let mut scan_range = ScanRange::full_table_scan();
-        let mut other_conds = groups.pop().unwrap();
         for i in 0..table_desc.order_column_indices().len() {
             let group = std::mem::take(&mut groups[i]);
             if group.is_empty() {
@@ -432,7 +537,12 @@ impl Condition {
                     scan_range.eq_conds.extend(eq_conds.into_iter());
                 }
                 0 => {
-                    scan_range.range = (lower_bound, upper_bound);
+                    let convert = |bound| match bound {
+                        Bound::Included(l) => Bound::Included(vec![Some(l)]),
+                        Bound::Excluded(l) => Bound::Excluded(vec![Some(l)]),
+                        Bound::Unbounded => Bound::Unbounded,
+                    };
+                    scan_range.range = (convert(lower_bound), convert(upper_bound));
                     other_conds.extend(groups[i + 1..].iter().flatten().cloned());
                     break;
                 }
diff --git a/src/storage/src/table/batch_table/storage_table.rs b/src/storage/src/table/batch_table/storage_table.rs
index ef3813dbaa15..c64498cbec9d 100644
--- a/src/storage/src/table/batch_table/storage_table.rs
+++ b/src/storage/src/table/batch_table/storage_table.rs
@@ -654,7 +654,6 @@ impl<S: StateStore, SD: ValueRowSerde> StorageTableInner<S, SD> {
     ) -> StorageResult<impl Stream<Item = StorageResult<KeyedRow<Bytes>>> + Send> {
         let start_key = self.serialize_pk_bound(&pk_prefix, range_bounds.start_bound(), true);
         let end_key = self.serialize_pk_bound(&pk_prefix, range_bounds.end_bound(), false);
-
         assert!(pk_prefix.len() <= self.pk_indices.len());
         let pk_prefix_indices = (0..pk_prefix.len())
             .map(|index| self.pk_indices[index])


Test patch:
diff --git a/src/frontend/planner_test/tests/testdata/input/row_filter.yaml b/src/frontend/planner_test/tests/testdata/input/row_filter.yaml
new file mode 100644
index 000000000000..2b1966f4316f
--- /dev/null
+++ b/src/frontend/planner_test/tests/testdata/input/row_filter.yaml
@@ -0,0 +1,25 @@
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v2,v3) > (1,2,3);
+  expected_outputs:
+  - batch_plan
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v2) > (1,2);
+  expected_outputs:
+  - batch_plan
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v3) > (2,3);
+  expected_outputs:
+  - batch_plan
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v3,v2,v1) > (1,2,3);
+  expected_outputs:
+  - batch_plan
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v2,v1) > (1,2,3);
+  expected_outputs:
+  - batch_plan
\ No newline at end of file
diff --git a/src/frontend/planner_test/tests/testdata/output/row_filter.yaml b/src/frontend/planner_test/tests/testdata/output/row_filter.yaml
new file mode 100644
index 000000000000..1ef80b8e025b
--- /dev/null
+++ b/src/frontend/planner_test/tests/testdata/output/row_filter.yaml
@@ -0,0 +1,34 @@
+# This file is automatically generated. See `src/frontend/planner_test/README.md` for more information.
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v2,v3) > (1,2,3);
+  batch_plan: |-
+    BatchExchange { order: [], dist: Single }
+    └─BatchScan { table: t, columns: [t.v1, t.v2, t.v3], scan_ranges: [(t.v1, t.v2, t.v3) > (Int32(1), Int32(2), Int32(3))], distribution: UpstreamHashShard(t.v1, t.v2, t.v3) }
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v2) > (1,2);
+  batch_plan: |-
+    BatchExchange { order: [], dist: Single }
+    └─BatchScan { table: t, columns: [t.v1, t.v2, t.v3], scan_ranges: [(t.v1, t.v2) > (Int32(1), Int32(2))], distribution: UpstreamHashShard(t.v1, t.v2, t.v3) }
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v3) > (2,3);
+  batch_plan: |-
+    BatchExchange { order: [], dist: Single }
+    └─BatchFilter { predicate: (Row(t.v1, t.v3) > '(2,3)':Struct(StructType { field_names: [], field_types: [Int32, Int32] })) }
+      └─BatchScan { table: t, columns: [t.v1, t.v2, t.v3], scan_ranges: [t.v1 > Int32(2)], distribution: UpstreamHashShard(t.v1, t.v2, t.v3) }
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v3,v2,v1) > (1,2,3);
+  batch_plan: |-
+    BatchExchange { order: [], dist: Single }
+    └─BatchFilter { predicate: (Row(t.v3, t.v2, t.v1) > '(1,2,3)':Struct(StructType { field_names: [], field_types: [Int32, Int32, Int32] })) }
+      └─BatchScan { table: t, columns: [t.v1, t.v2, t.v3], distribution: UpstreamHashShard(t.v1, t.v2, t.v3) }
+- sql: |
+    create table t(v1 int, v2 int, v3 int, primary key(v1,v2,v3));
+    select * from t where (v1,v2,v1) > (1,2,3);
+  batch_plan: |-
+    BatchExchange { order: [], dist: Single }
+    └─BatchFilter { predicate: (Row(t.v1, t.v2, t.v1) > '(1,2,3)':Struct(StructType { field_names: [], field_types: [Int32, Int32, Int32] })) }
+      └─BatchScan { table: t, columns: [t.v1, t.v2, t.v3], scan_ranges: [(t.v1, t.v2) > (Int32(1), Int32(2))], distribution: UpstreamHashShard(t.v1, t.v2, t.v3) }