diff --git a/src/common/src/row/owned_row.rs b/src/common/src/row/owned_row.rs
index 983a96b91a5a..8ad7c332ecad 100644
--- a/src/common/src/row/owned_row.rs
+++ b/src/common/src/row/owned_row.rs
@@ -89,75 +89,6 @@ impl OwnedRow {
     }
 }
 
-impl OwnedRow {
-    pub fn is_null(&self, idx: usize) -> bool {
-        self[idx].is_none()
-    }
-
-    pub fn get_int16(&self, idx: usize) -> i16 {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Int16(num) => *num,
-            _ => unreachable!("type is not int16 at index: {}", idx),
-        }
-    }
-
-    pub fn get_int32(&self, idx: usize) -> i32 {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Int32(num) => *num,
-            _ => unreachable!("type is not int32 at index: {}", idx),
-        }
-    }
-
-    pub fn get_int64(&self, idx: usize) -> i64 {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Int64(num) => *num,
-            _ => unreachable!("type is not int64 at index: {}", idx),
-        }
-    }
-
-    pub fn get_f32(&self, idx: usize) -> f32 {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Float32(num) => num.into_inner(),
-            _ => unreachable!("type is not float32 at index: {}", idx),
-        }
-    }
-
-    pub fn get_f64(&self, idx: usize) -> f64 {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Float64(num) => num.into_inner(),
-            _ => unreachable!("type is not float64 at index: {}", idx),
-        }
-    }
-
-    pub fn get_bool(&self, idx: usize) -> bool {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Bool(num) => *num,
-            _ => unreachable!("type is not boolean at index: {}", idx),
-        }
-    }
-
-    pub fn get_utf8(&self, idx: usize) -> &str {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Utf8(s) => s.as_ref(),
-            _ => unreachable!("type is not utf8 at index: {}", idx),
-        }
-    }
-
-    pub fn get_datetime(&self, idx: usize) -> &Timestamp {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Timestamp(dt) => dt,
-            _ => unreachable!("type is not NaiveDateTime at index: {}", idx),
-        }
-    }
-
-    pub fn get_decimal(&self, idx: usize) -> &Decimal {
-        match self[idx].as_ref().unwrap() {
-            ScalarImpl::Decimal(d) => d,
-            _ => unreachable!("type is not NaiveDateTime at index: {}", idx),
-        }
-    }
-}
-
 impl EstimateSize for OwnedRow {
     fn estimated_heap_size(&self) -> usize {
         // FIXME(bugen): this is not accurate now as the heap size of some `Scalar` is not counted.
diff --git a/src/java_binding/src/lib.rs b/src/java_binding/src/lib.rs
index 955f2f6e1749..904ef025a300 100644
--- a/src/java_binding/src/lib.rs
+++ b/src/java_binding/src/lib.rs
@@ -35,7 +35,7 @@ use once_cell::sync::OnceCell;
 use prost::{DecodeError, Message};
 use risingwave_common::array::{ArrayError, StreamChunk};
 use risingwave_common::hash::VirtualNode;
-use risingwave_common::row::OwnedRow;
+use risingwave_common::row::{OwnedRow, Row};
 use risingwave_storage::error::StorageError;
 use thiserror::Error;
 use tokio::runtime::Runtime;
@@ -388,7 +388,7 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowIsNull<'a>(
     idx: jint,
 ) -> jboolean {
     execute_and_catch(env, move || {
-        Ok(pointer.as_ref().is_null(idx as usize) as jboolean)
+        Ok(pointer.as_ref().datum_at(idx as usize).is_none() as jboolean)
     })
 }
 
@@ -398,7 +398,13 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetInt16Value
     pointer: Pointer<'a, JavaBindingRow>,
     idx: jint,
 ) -> jshort {
-    execute_and_catch(env, move || Ok(pointer.as_ref().get_int16(idx as usize)))
+    execute_and_catch(env, move || {
+        Ok(pointer
+            .as_ref()
+            .datum_at(idx as usize)
+            .unwrap()
+            .into_int16())
+    })
 }
 
 #[no_mangle]
@@ -407,7 +413,13 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetInt32Value
     pointer: Pointer<'a, JavaBindingRow>,
     idx: jint,
 ) -> jint {
-    execute_and_catch(env, move || Ok(pointer.as_ref().get_int32(idx as usize)))
+    execute_and_catch(env, move || {
+        Ok(pointer
+            .as_ref()
+            .datum_at(idx as usize)
+            .unwrap()
+            .into_int32())
+    })
 }
 
 #[no_mangle]
@@ -416,7 +428,13 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetInt64Value
     pointer: Pointer<'a, JavaBindingRow>,
     idx: jint,
 ) -> jlong {
-    execute_and_catch(env, move || Ok(pointer.as_ref().get_int64(idx as usize)))
+    execute_and_catch(env, move || {
+        Ok(pointer
+            .as_ref()
+            .datum_at(idx as usize)
+            .unwrap()
+            .into_int64())
+    })
 }
 
 #[no_mangle]
@@ -425,7 +443,14 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetFloatValue
     pointer: Pointer<'a, JavaBindingRow>,
     idx: jint,
 ) -> jfloat {
-    execute_and_catch(env, move || Ok(pointer.as_ref().get_f32(idx as usize)))
+    execute_and_catch(env, move || {
+        Ok(pointer
+            .as_ref()
+            .datum_at(idx as usize)
+            .unwrap()
+            .into_float32()
+            .into())
+    })
 }
 
 #[no_mangle]
@@ -434,7 +459,14 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetDoubleValu
     pointer: Pointer<'a, JavaBindingRow>,
     idx: jint,
 ) -> jdouble {
-    execute_and_catch(env, move || Ok(pointer.as_ref().get_f64(idx as usize)))
+    execute_and_catch(env, move || {
+        Ok(pointer
+            .as_ref()
+            .datum_at(idx as usize)
+            .unwrap()
+            .into_float64()
+            .into())
+    })
 }
 
 #[no_mangle]
@@ -444,7 +476,7 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetBooleanVal
     idx: jint,
 ) -> jboolean {
     execute_and_catch(env, move || {
-        Ok(pointer.as_ref().get_bool(idx as usize) as jboolean)
+        Ok(pointer.as_ref().datum_at(idx as usize).unwrap().into_bool() as jboolean)
     })
 }
 
@@ -455,7 +487,7 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetStringValu
     idx: jint,
 ) -> JString<'a> {
     execute_and_catch(env, move || {
-        Ok(env.new_string(pointer.as_ref().get_utf8(idx as usize))?)
+        Ok(env.new_string(pointer.as_ref().datum_at(idx as usize).unwrap().into_utf8())?)
     })
 }
 
@@ -468,7 +500,9 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetTimestampV
     execute_and_catch(env, move || {
         let millis = pointer
             .as_ref()
-            .get_datetime(idx as usize)
+            .datum_at(idx as usize)
+            .unwrap()
+            .into_timestamp()
             .0
             .timestamp_millis();
         let (ts_class_ref, constructor) = pointer
@@ -494,7 +528,12 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_rowGetDecimalVal
     idx: jint,
 ) -> JObject<'a> {
     execute_and_catch(env, move || {
-        let value = pointer.as_ref().get_decimal(idx as usize).to_string();
+        let value = pointer
+            .as_ref()
+            .datum_at(idx as usize)
+            .unwrap()
+            .into_decimal()
+            .to_string();
         let string_value = env.new_string(value)?;
         let (decimal_class_ref, constructor) = pointer
             .as_ref()