fix(batch): fix batch range scan for null (#13524)

e2e_test/batch/basic/null_range_scan.slt.part

@@ -0,0 +1,20 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+CREATE TABLE t0(c0 INT, c1 INT, PRIMARY KEY(c1, c0));
+
+statement ok
+INSERT INTO t0(c0) VALUES (1);
+
+query II rowsort
+SELECT * FROM t0 WHERE (c1 > 1) IS NULL;
+----
+1 NULL
+
+query II rowsort
+SELECT * FROM t0 WHERE (c1 > 1);
+----
+
+statement ok
+drop table t0;

src/batch/src/executor/row_seq_scan.rs

@@ -398,22 +398,49 @@ impl<S: StateStore> RowSeqScanExecutor<S> {
             next_col_bounds,
         } = scan_range;
 
-        let (start_bound, end_bound) =
+        let (start_bound, end_bound, null_is_largest) =
             if table.pk_serializer().get_order_types()[pk_prefix.len()].is_ascending() {
-                (next_col_bounds.0, next_col_bounds.1)
+                (next_col_bounds.0, next_col_bounds.1, true)
             } else {
-                (next_col_bounds.1, next_col_bounds.0)
+                (next_col_bounds.1, next_col_bounds.0, false)
             };
 
+        let start_bound_is_bounded = !matches!(start_bound, Bound::Unbounded);
+        let end_bound_is_bounded = !matches!(end_bound, Bound::Unbounded);
+
         // Range Scan.
         assert!(pk_prefix.len() < table.pk_indices().len());
         let iter = table
             .batch_iter_with_pk_bounds(
                 epoch.into(),
                 &pk_prefix,
                 (
-                    start_bound.map(|x| OwnedRow::new(vec![x])),
-                    end_bound.map(|x| OwnedRow::new(vec![x])),
+                    match start_bound {
+                        Bound::Unbounded => {
+                            if end_bound_is_bounded && !null_is_largest {
+                                // Since Null is the smallest value, we need to skip it for range scan to meet the 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 && null_is_largest {
+                                // Since Null is the largest value, we need to skip it for range scan to meet the 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])),
+                    },
                 ),
                 ordered,
                 PrefetchOptions::new_for_large_range_scan(),