[tools] Add --fault_tolerant to 'kudu table copy/scan'

Add --fault_tolerant to 'kudu table copy/scan' to make
the results are returned in primary key order and make
the scanner fault-tolerant.

Change-Id: I1a05792db9dceb5e774ce1602158bb636bba04d0

src/kudu/client/scan_configuration.cc

@@ -168,6 +168,8 @@ Status ScanConfiguration::SetReadMode(KuduScanner::ReadMode read_mode) {
 
 Status ScanConfiguration::SetFaultTolerant(bool fault_tolerant) {
   if (fault_tolerant) {
+    // TODO(yingchun): this will overwrite the user set read mode, maybe it
+    //  should return error if there is any conflict.
     RETURN_NOT_OK(SetReadMode(KuduScanner::READ_AT_SNAPSHOT));
   }
   is_fault_tolerant_ = fault_tolerant;

src/kudu/tools/table_scanner.cc

@@ -98,6 +98,10 @@ DEFINE_string(create_table_hash_bucket_nums, "",
               "The number of hash buckets in each hash dimension seperated by comma");
 DEFINE_bool(fill_cache, true,
             "Whether to fill block cache when scanning.");
+DEFINE_bool(fault_tolerant, false,
+            "Whether the results are returned in primary key order, make the scanner "
+            "fault-tolerant. This also means the scanner can recover if the server it "
+            "is scanning fails in the middle of a scan.");
 DEFINE_string(predicates, "",
               "Query predicates on columns. Unlike traditional SQL syntax, "
               "the scan tool's simple query predicates are represented in a "
@@ -668,6 +672,14 @@ Status TableScanner::StartWork(WorkType type) {
   }
   RETURN_NOT_OK(builder.SetSelection(replica_selection_));
   RETURN_NOT_OK(builder.SetTimeoutMillis(FLAGS_timeout_ms));
+  if (FLAGS_fault_tolerant) {
+    // TODO(yingchun): push down this judgement to ScanConfiguration::SetFaultTolerant
+    if (mode_ && *mode_ != KuduScanner::READ_AT_SNAPSHOT) {
+      return Status::InvalidArgument(Substitute("--fault_tolerant is conflict with "
+          "the non-READ_AT_SNAPSHOT read mode"));
+    }
+    RETURN_NOT_OK(builder.SetFaultTolerant());
+  }
 
   // Set projection if needed.
   if (type == WorkType::kScan) {

src/kudu/tools/tool_action_table.cc

@@ -160,6 +160,7 @@ DEFINE_bool(show_avro_format_schema, false,
             "partition/owner/comments. It cannot be used in conjunction with other flags");
 
 DECLARE_bool(create_table);
+DECLARE_bool(fault_tolerant);
 DECLARE_int32(create_table_replication_factor);
 DECLARE_bool(row_count_only);
 DECLARE_bool(show_scanner_stats);