fix(storage): support excluded bound in storage table

e2e_test/batch/issue_16219.slt

@@ -0,0 +1,54 @@
+# This is a test on a panic in select with excluded upper bound: https://github.com/risingwavelabs/risingwave/issues/16219
+
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+create table t (a varchar);
+
+statement ok
+create materialized view mv as select * from t order by a desc;
+
+query T
+select * from mv where a < '';
+----
+
+query T
+select * from mv where a > '';
+----
+
+query T
+select * from mv where a <= '';
+----
+
+query T
+select * from mv where a >= '';
+----
+
+statement ok
+INSERT INTO t VALUES ('foo');
+
+query T
+select * from mv where a < '';
+----
+
+query T
+select * from mv where a > '';
+----
+foo
+
+query T
+select * from mv where a <= '';
+----
+
+
+query T
+select * from mv where a >= '';
+----
+foo
+
+statement ok
+drop materialized view mv;
+
+statement ok
+drop table t;

src/storage/src/table/batch_table/storage_table.rs

@@ -717,17 +717,7 @@ impl<S: StateStore, SD: ValueRowSerde> StorageTableInner<S, SD> {
                 let pk_prefix_serializer = self.pk_serializer.prefix(pk_prefix.len() + k.len());
                 let key = pk_prefix.chain(k);
                 let serialized_key = serialize_pk(&key, &pk_prefix_serializer);
-                if is_start_bound {
-                    // Storage doesn't support excluded begin key yet, so transform it to
-                    // included.
-                    // We always serialize a u8 for null of datum which is not equal to '\xff',
-                    // so we can assert that the next_key would never be empty.
-                    let next_serialized_key = next_key(&serialized_key);
-                    assert!(!next_serialized_key.is_empty());
-                    Included(Bytes::from(next_serialized_key))
-                } else {
-                    Excluded(serialized_key)
-                }
+                Excluded(serialized_key)
             }
             Unbounded => {
                 let pk_prefix_serializer = self.pk_serializer.prefix(pk_prefix.len());