close #11384

Signed-off-by: qupeng <qupeng@pingcap.com>
tikv · Nov 30, 2021 · bf36779 · bf36779
1 parent 2bbdc8d
commit bf36779
Show file tree

Hide file tree

Showing 19 changed files with 569 additions and 140 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/components/backup/src/writer.rs b/components/backup/src/writer.rs
@@ -425,6 +425,7 @@ mod tests {
     use std::path::Path;
     use tempfile::TempDir;
     use tikv::storage::TestEngineBuilder;
+    use txn_types::OldValue;
 
     type CfKvs<'a> = (engine_traits::CfName, &'a [(&'a [u8], &'a [u8])]);
 
@@ -522,7 +523,7 @@ mod tests {
                 vec![TxnEntry::Commit {
                     default: (vec![], vec![]),
                     write: (vec![b'a'], vec![b'a']),
-                    old_value: None,
+                    old_value: OldValue::None,
                 }]
                 .into_iter(),
                 false,
@@ -562,12 +563,12 @@ mod tests {
                     TxnEntry::Commit {
                         default: (vec![b'a'], vec![b'a']),
                         write: (vec![b'a'], vec![b'a']),
-                        old_value: None,
+                        old_value: OldValue::None,
                     },
                     TxnEntry::Commit {
                         default: (vec![], vec![]),
                         write: (vec![b'b'], vec![]),
-                        old_value: None,
+                        old_value: OldValue::None,
                     },
                 ]
                 .into_iter(),

diff --git a/components/cdc/Cargo.toml b/components/cdc/Cargo.toml
@@ -24,9 +24,11 @@ failpoints = ["tikv/failpoints"]
 [dependencies]
 bitflags = "1.0"
 crossbeam = "0.8"
+engine_rocks = { path = "../engine_rocks", default-features = false }
 engine_traits = { path = "../engine_traits", default-features = false }
 futures = "0.3"
 grpcio = { version = "0.9",  default-features = false, features = ["openssl-vendored", "protobuf-codec"] }
+keys = { path = "../keys" }
 kvproto = { git = "https://github.com/pingcap/kvproto.git" }
 pd_client = { path = "../pd_client", default-features = false }
 raft = { version = "0.6.0-alpha", default-features = false, features = ["protobuf-codec"] }

diff --git a/components/cdc/src/delegate.rs b/components/cdc/src/delegate.rs
@@ -453,7 +453,7 @@ impl Delegate {
                         current_rows_size = 0;
                     }
                     current_rows_size += row_size;
-                    row.old_value = old_value.unwrap_or_default();
+                    row.old_value = old_value.finalized().unwrap_or_default();
                     rows.last_mut().unwrap().push(row);
                 }
                 Some(TxnEntry::Commit {
@@ -480,7 +480,7 @@ impl Delegate {
                         continue;
                     }
                     set_event_row_type(&mut row, EventLogType::Committed);
-                    row.old_value = old_value.unwrap_or_default();
+                    row.old_value = old_value.finalized().unwrap_or_default();
                     let row_size = row.key.len() + row.value.len();
                     if current_rows_size + row_size >= CDC_EVENT_MAX_BYTES {
                         rows.push(Vec::with_capacity(entries_len));
@@ -535,13 +535,7 @@ impl Delegate {
                     .with_label_values(&["all"])
                     .observe(start.saturating_elapsed().as_secs_f64());
                 if let Some(statistics) = statistics {
-                    for (cf, cf_details) in statistics.details().iter() {
-                        for (tag, count) in cf_details.iter() {
-                            CDC_OLD_VALUE_SCAN_DETAILS
-                                .with_label_values(&[*cf, *tag])
-                                .inc_by(*count as u64);
-                        }
-                    }
+                    flush_oldvalue_stats(&statistics, TAG_DELTA_CHANGE);
                 }
             }
         };