fix(common): order null as largest by default to align with PostgreSQL (risingwavelabs#4263)

* encode and decode null as largest in memcomparable encoding

* fix unit test except lookup join

```
NEXTEST_EXPERIMENTAL_FILTER_EXPR=1 ./risedev test -E 'not test(~lookup_join)'
```

* update lookup join unit test

* update e2e

* update batch ordering of null

* update e2e

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

Author: 2 people

e2e_test/batch/basic/order_by.slt.part

@@ -91,17 +91,17 @@ insert into t values (null, 7, null);
 query III
 select * from t order by v1 limit 2;
 ----
-NULL 7 NULL
 1 4 2
+2 3 3
 
 query III
 select * from t order by v1 desc limit 7;
 ----
+NULL 7 NULL
 4 3 5
 3 4 4
 2 3 3
 1 4 2
-NULL 7 NULL
 
 statement ok
 drop table t;

e2e_test/batch/basic/subquery.slt.part

@@ -27,12 +27,12 @@ select * from t1 where exists(select * from t2 where t1.x = t2.x and t1.y <> t2.
 query III
 select * from t1 where not exists(select * from t2 where t1.x = t2.x and t1.y <> t2.y) order by t1.x, t1.y
 ----
-NULL NULL
-NULL 1
-NULL 2
 1 NULL
-2 NULL
 2 2
+2 NULL
+NULL 1
+NULL 2
+NULL NULL
 
 query III
 select * from t1 where t1.y in (select t1.y from t2 where t1.x = t2.x) order by t1.x, t1.y
@@ -43,16 +43,16 @@ select * from t1 where t1.y in (select t1.y from t2 where t1.x = t2.x) order by
 query III
 select * from t1 where exists(select x from t2 where t1.x = t2.x and t2.y in (select t3.y from t3 where t1.x = t3.x)) order by t1.x, t1.y
 ----
-1 NULL
 1 1
-2 NULL
+1 NULL
 2 2
+2 NULL
 
 query III
 select * from t1 where exists(select t2.x from t2 join t3 on t2.x = t3.x and t1.y = t2.y and t1.y = t3.y) order by t1.x, t1.y
 ----
-NULL 2
 2 2
+NULL 2
 
 
 statement ok
@@ -62,4 +62,4 @@ statement ok
 drop table t2;
 
 statement ok
-drop table t3;
+drop table t3;

src/batch/src/executor/join/lookup_join.rs

@@ -835,7 +835,6 @@ mod tests {
     async fn test_left_outer_join() {
         let expected = DataChunk::from_pretty(
             "i f   i f
-             . .   . .
              1 6.1 1 9.2
              2 5.5 2 4.4
              2 5.5 2 5.5
@@ -845,7 +844,8 @@ mod tests {
              5 4.1 5 3.7
              5 4.1 5 2.3
              5 9.1 5 3.7
-             5 9.1 5 2.3",
+             5 9.1 5 2.3
+             . .   . .",
         );
 
         do_test(JoinType::LeftOuter, None, expected).await;
@@ -869,8 +869,8 @@ mod tests {
     async fn test_left_anti_join() {
         let expected = DataChunk::from_pretty(
             "i f
-             . .
-             3 3.9",
+             3 3.9
+             . .",
         );
 
         do_test(JoinType::LeftAnti, None, expected).await;
@@ -902,13 +902,13 @@ mod tests {
     async fn test_left_outer_join_with_condition() {
         let expected = DataChunk::from_pretty(
             "i f   i f
-             . .   . .
              1 6.1 1 9.2
              2 5.5 2 5.5
              2 8.4 2 5.5
              3 3.9 . .
              5 4.1 . .
-             5 9.1 . .",
+             5 9.1 . .
+             . .   . .",
         );
 
         let condition = Some(new_binary_expr(
@@ -950,10 +950,10 @@ mod tests {
     async fn test_left_anti_join_with_condition() {
         let expected = DataChunk::from_pretty(
             "i f
-            . .
             3 3.9
             5 4.1
-            5 9.1",
+            5 9.1
+            . .",
         );
 
         let condition = Some(new_binary_expr(

src/common/src/types/mod.rs

@@ -359,11 +359,12 @@ pub fn serialize_datum_ref_into(
     datum_ref: &DatumRef,
     serializer: &mut memcomparable::Serializer<impl BufMut>,
 ) -> memcomparable::Result<()> {
+    // By default, `null` is treated as largest in PostgreSQL.
     if let Some(datum_ref) = datum_ref {
-        1u8.serialize(&mut *serializer)?;
+        0u8.serialize(&mut *serializer)?;
         datum_ref.serialize(serializer)?;
     } else {
-        0u8.serialize(serializer)?;
+        1u8.serialize(serializer)?;
     }
     Ok(())
 }
@@ -384,11 +385,12 @@ pub fn serialize_datum_into(
     datum: &Datum,
     serializer: &mut memcomparable::Serializer<impl BufMut>,
 ) -> memcomparable::Result<()> {
+    // By default, `null` is treated as largest in PostgreSQL.
     if let Some(datum) = datum {
-        1u8.serialize(&mut *serializer)?;
+        0u8.serialize(&mut *serializer)?;
         datum.serialize(serializer)?;
     } else {
-        0u8.serialize(serializer)?;
+        1u8.serialize(serializer)?;
     }
     Ok(())
 }
@@ -411,8 +413,8 @@ pub fn deserialize_datum_from(
 ) -> memcomparable::Result<Datum> {
     let null_tag = u8::deserialize(&mut *deserializer)?;
     match null_tag {
-        0 => Ok(None),
-        1 => Ok(Some(ScalarImpl::deserialize(ty.clone(), deserializer)?)),
+        1 => Ok(None),
+        0 => Ok(Some(ScalarImpl::deserialize(ty.clone(), deserializer)?)),
         _ => Err(memcomparable::Error::InvalidTagEncoding(null_tag as _)),
     }
 }
@@ -757,8 +759,8 @@ impl ScalarImpl {
         let base_position = deserializer.position();
         let null_tag = u8::deserialize(&mut *deserializer)?;
         match null_tag {
-            0 => {}
-            1 => {
+            1 => {}
+            0 => {
                 use std::mem::size_of;
                 let len = match data_type {
                     DataType::Int16 => size_of::<i16>(),

src/common/src/util/ordered/serde.rs

@@ -204,8 +204,8 @@ mod tests {
         array.sort();
         // option 1 byte || number 2 bytes
         assert_eq!(array[0][2], !6i16.to_be_bytes()[1]);
-        assert_eq!(&array[1][3..], [1, 1, b'a', b'b', b'c', 0, 0, 0, 0, 0, 3u8]);
-        assert_eq!(&array[2][3..], [1, 1, b'a', b'b', b'd', 0, 0, 0, 0, 0, 3u8]);
+        assert_eq!(&array[1][3..], [0, 1, b'a', b'b', b'c', 0, 0, 0, 0, 0, 3u8]);
+        assert_eq!(&array[2][3..], [0, 1, b'a', b'b', b'd', 0, 0, 0, 0, 0, 3u8]);
     }
 
     #[test]
@@ -249,7 +249,7 @@ mod tests {
         assert_eq!(
             array[0][..11],
             [
-                !(1u8),
+                !(0u8),
                 !(1u8),
                 !(b'a'),
                 !(b'b'),
@@ -262,8 +262,8 @@ mod tests {
                 !(3u8)
             ]
         );
-        assert_eq!(array[1][11..], [1, 1, b'j', b'm', b'z', 0, 0, 0, 0, 0, 3u8]);
-        assert_eq!(array[2][11..], [1, 1, b'm', b'j', b'z', 0, 0, 0, 0, 0, 3u8]);
+        assert_eq!(array[1][11..], [0, 1, b'j', b'm', b'z', 0, 0, 0, 0, 0, 3u8]);
+        assert_eq!(array[2][11..], [0, 1, b'm', b'j', b'z', 0, 0, 0, 0, 0, 3u8]);
     }
 
     #[test]

src/common/src/util/sort_util.rs

@@ -172,11 +172,8 @@ where
         (Some(l), Some(r)) => l.cmp(r),
         (None, None) => Ordering::Equal,
         // TODO(yuchao): `null first` / `null last` is not supported yet.
-        // To be consistent with memcomparable (#116) encoding, `null` is treated as less than any
-        // non-null value. This is contrary to PostgreSQL's default behavior, where `null`
-        // is treated as largest.
-        (Some(_), None) => Ordering::Greater,
-        (None, Some(_)) => Ordering::Less,
+        (Some(_), None) => Ordering::Less,
+        (None, Some(_)) => Ordering::Greater,
     };
     if *order_type == OrderType::Descending {
         ord.reverse()