| 2369 | } |
| 2370 | |
| 2371 | int ScanAndDiffData(Table* src, Table* dest, const std::string& src_table_name, |
| 2372 | const std::string& dest_table_name, const std::string& tablet_id, |
| 2373 | const std::string& start_key, const std::string& end_key, |
| 2374 | const std::map<std::string, std::string>& tables_cf_map, |
| 2375 | const std::map<std::string, std::string>& tables_cf_version_map, |
| 2376 | DiffStatData* diff_stat_data) { |
| 2377 | int res = 0; |
| 2378 | |
| 2379 | diff_stat_data->reset(); |
| 2380 | |
| 2381 | std::vector<std::string> cfs; |
| 2382 | auto it = tables_cf_map.find(src_table_name); |
| 2383 | if (it != tables_cf_map.cend()) { |
| 2384 | std::string delimiter(FLAGS_lg_and_cf_delimiter); |
| 2385 | SplitString(it->second, delimiter, &cfs); |
| 2386 | } |
| 2387 | |
| 2388 | ResultStream* src_result_stream = NULL; |
| 2389 | ResultStream* dest_result_stream = NULL; |
| 2390 | if (StartScan(src, src_table_name, start_key, end_key, cfs, "src", &src_result_stream) < 0) { |
| 2391 | return -1; |
| 2392 | } |
| 2393 | if (StartScan(dest, dest_table_name, start_key, end_key, cfs, "dest", &dest_result_stream) < 0) { |
| 2394 | delete src_result_stream; |
| 2395 | return -1; |
| 2396 | } |
| 2397 | |
| 2398 | std::ostringstream diffdata_only_in_src, diffdata_only_in_dest, diffdata_both_have_but_diff; |
| 2399 | |
| 2400 | CompareContext ctx(src_result_stream, dest_result_stream, diffdata_only_in_src, |
| 2401 | diffdata_only_in_dest, diffdata_both_have_but_diff, diff_stat_data, |
| 2402 | src_table_name, tables_cf_version_map, dest); |
| 2403 | ctx.counter.Inc(); |
| 2404 | uint64_t cnt = 0; |
| 2405 | while (true) { |
| 2406 | bool src_done = false; |
| 2407 | bool dest_done = false; |
| 2408 | |
| 2409 | if (FLAGS_diff_scan_count_per_interval > 0) { |
| 2410 | cnt++; |
| 2411 | if (cnt % FLAGS_diff_scan_count_per_interval == 0) { |
| 2412 | ThisThread::Sleep(FLAGS_diff_scan_interval_ns); |
| 2413 | } |
| 2414 | } |
| 2415 | |
| 2416 | if (IsScanFinish(src_table_name, dest_table_name, start_key, end_key, src_result_stream, |
| 2417 | dest_result_stream, &src_done, &dest_done, &res)) { |
| 2418 | break; |
| 2419 | } |
| 2420 | |
| 2421 | diff_stat_data->in_src_or_in_dest++; |
| 2422 | if (src_done && !dest_done) { |
| 2423 | VLOG(1) << "dest: [" << dest_result_stream->RowName() << "] [" << dest_result_stream->Family() |
| 2424 | << "] [" << dest_result_stream->Qualifier() << "] [" |
| 2425 | << dest_result_stream->Timestamp() << "]"; |
| 2426 | DestRecordAndNext(ctx); |
| 2427 | VLOG(1) << "[diff] src_done"; |
| 2428 | } else if (!src_done && dest_done) { |
no test coverage detected