fix(meta): fix list changed log epoch retrun null epoch (#18273)

src/frontend/src/session/cursor_manager.rs

@@ -320,7 +320,7 @@ impl SubscriptionCursor {
                                 &mut chunk_stream,
                                 formats,
                                 &from_snapshot,
-                                &self.fields,
+                                &fields,
                                 handle_args.session.clone(),
                             );
 

src/meta/src/hummock/manager/versioning.rs

@@ -310,7 +310,7 @@ impl HummockManager {
             .get(&TableId::new(table_id))
         {
             let table_change_log = table_change_log.clone();
-            table_change_log.get_epochs(min_epoch, max_count as usize)
+            table_change_log.get_non_empty_epochs(min_epoch, max_count as usize)
         } else {
             vec![]
         }

src/storage/hummock_sdk/src/change_log.rs

@@ -90,9 +90,16 @@ impl TableChangeLog {
         &self.0[start..end]
     }
 
-    pub fn get_epochs(&self, min_epoch: u64, max_count: usize) -> Vec<u64> {
+    /// Returns epochs where value is non-null and >= `min_epoch`.
+    pub fn get_non_empty_epochs(&self, min_epoch: u64, max_count: usize) -> Vec<u64> {
         self.filter_epoch((min_epoch, u64::MAX))
             .iter()
+            .filter(|epoch_change_log| {
+                // Filter out empty change logs
+                let new_value_empty = epoch_change_log.new_value.is_empty();
+                let old_value_empty = epoch_change_log.old_value.is_empty();
+                !new_value_empty || !old_value_empty
+            })
             .flat_map(|epoch_change_log| epoch_change_log.epochs.iter().cloned())
             .filter(|a| a >= &min_epoch)
             .clone()