diff --git a/e2e_test/batch/issue_16219.slt b/e2e_test/batch/issue_16219.slt
new file mode 100644
index 0000000000000..aea23019831e9
--- /dev/null
+++ b/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;
\ No newline at end of file
diff --git a/src/storage/src/table/batch_table/storage_table.rs b/src/storage/src/table/batch_table/storage_table.rs
index acfd4ffcb9bb1..6fc4e6bcce865 100644
--- a/src/storage/src/table/batch_table/storage_table.rs
+++ b/src/storage/src/table/batch_table/storage_table.rs
@@ -37,7 +37,7 @@ use risingwave_common::util::sort_util::OrderType;
 use risingwave_common::util::value_encoding::column_aware_row_encoding::ColumnAwareSerde;
 use risingwave_common::util::value_encoding::{BasicSerde, EitherSerde};
 use risingwave_hummock_sdk::key::{
-    end_bound_of_prefix, next_key, prefixed_range_with_vnode, CopyFromSlice, TableKeyRange,
+    end_bound_of_prefix, prefixed_range_with_vnode, CopyFromSlice, TableKeyRange,
 };
 use risingwave_hummock_sdk::HummockReadEpoch;
 use risingwave_pb::plan_common::StorageTableDesc;
@@ -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());