<commit_before>// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//

#include <stdlib.h>
#include <unistd.h>

#include <fstream>
#include <iostream>
#include <limits>
#include <sstream>

#include <gflags/gflags.h>
#include <glog/logging.h>

#include "common/base/string_ext.h"
#include "common/base/string_number.h"
#include "common/file/file_path.h"
#include "io/coding.h"
#include "proto/kv_helper.h"
#include "proto/proto_helper.h"
#include "proto/tabletnode.pb.h"
#include "proto/tabletnode_client.h"
#include "sdk/client_impl.h"
#include "sdk/sdk_utils.h"
#include "sdk/sdk_zk.h"
#include "sdk/table_impl.h"
#include "sdk/tera.h"
#include "utils/string_util.h"
#include "utils/tprinter.h"
#include "utils/utils_cmd.h"
#include "version.h"

DECLARE_string(flagfile);
DECLARE_string(log_dir);
DECLARE_string(tera_master_meta_table_name);
DECLARE_string(tera_zk_addr_list);
DECLARE_string(tera_zk_root_path);

DEFINE_int32(tera_client_batch_put_num, 1000, "num of each batch in batch put mode");
DEFINE_int32(tera_client_scan_package_size, 1024, "the package size (in KB) of each scan request");
DEFINE_bool(tera_client_scan_async_enabled, false, "enable the streaming scan mode");

volatile int32_t g_start_time = 0;
volatile int32_t g_end_time = 0;
volatile int32_t g_used_time = 0;
volatile int32_t g_last_time  = 0;
volatile int64_t g_total_size = 0;
volatile int32_t g_key_num = 0;
Mutex g_stat_lock;

volatile int32_t g_cur_batch_num = 0;

using namespace tera;

void Usage(const std::string& prg_name) {
    std::cout << "\nSYNOPSIS\n";
    std::cout << "       " << prg_name << "  OPERATION  [OPTION...] \n\n";
    std::cout << "DESCRIPTION \n\
       create   <schema> [<delimiter_file>]                                 \n\
              - schema syntax (all properties are optional):                \n\
                    tablename <rawkey=binary,splitsize=1024,...> {          \n\
                        lg1 <storage=flash,...> {                           \n\
                            cf1 <maxversion=3>,                             \n\
                            cf2...},                                        \n\
                        lg2...                                              \n\
                    }                                                       \n\
              - kv mode schema:                                             \n\
                    tablename <splitsize=1024, storage=memory, ...>         \n\
              - simple mode schema:                                         \n\
                    tablename{cf1, cf2, cf3, ...}                           \n\
                                                                            \n\
       createbyfile   <schema_file> [<delimiter_file>]                      \n\
                                                                            \n\
       update <schema>                                                      \n\
              - kv schema:                                                  \n\
                    e.g. tablename<splitsize=512,storage=disk>              \n\
              - table schema:                                               \n\
                - update properties                                         \n\
                    e.g. tablename<splitsize=512>                           \n\
                    e.g. tablename{lg0{cf0<ttl=123>}}                       \n\
                    e.g. tablename<mergesize=233>{lg0<storage=disk>{cf0<ttl=123>}}       \n\
                - add new cf                                                \n\
                    e.g. tablename{lg0{cf0<ttl=250>,new_cf<op=add,ttl=233>}}\n\
                - delete cf                                                 \n\
                    e.g. tablename{lg0{cf0<op=del>}}                        \n\
                                                                            \n\
       enable/disable/drop  <tablename>                                     \n\
                                                                            \n\
       put      <tablename> <rowkey> [<columnfamily:qualifier>] <value>     \n\
                                                                            \n\
       put-ttl  <tablename> <rowkey> [<columnfamily:qualifier>] <value> <ttl(second)>    \n\
                                                                            \n\
       putif    <tablename> <rowkey> [<columnfamily:qualifier>] <value>     \n\
                                                                            \n\
       get      <tablename> <rowkey> [<columnfamily:qualifier>]             \n\
                                                                            \n\
       scan[allv] <tablename> <startkey> <endkey> [<\"cf1|cf2\">]           \n\
                scan table from startkey to endkey.                         \n\
                (return all qulifier version when using suffix \"allv\")    \n\
                                                                            \n\
       delete[1v] <tablename> <rowkey> [<columnfamily:qualifier>]           \n\
                delete row/columnfamily/qualifiers.                         \n\
                (only delete latest version when using suffix \"1v\")       \n\
                                                                            \n\
       put_counter <tablename> <rowkey> [<columnfamily:qualifier>] <integer>\n\
                                                                            \n\
       get_counter <tablename> <rowkey> [<columnfamily:qualifier>]          \n\
                                                                            \n\
       add      <tablename> <rowkey> <columnfamily:qualifier>   delta       \n\
                add 'delta'(int64_t) to specified cell                      \n\
       append   <tablename> <rowkey> [<columnfamily:qualifier>] <value>     \n\
                                                                            \n\
       batchput <tablename> <input file>                                    \n\
                                                                            \n\
       batchget <tablename> <input file>                                    \n\
                                                                            \n\
       show[x]  [<tablename>]                                               \n\
                show table list or tablets info.                            \n\
                (show more detail when using suffix \"x\")                  \n\
                                                                            \n\
       showschema[x] <tablename>                                            \n\
                show table schema (show more detail when using suffix \"x\")\n\
                                                                            \n\
       showts[x] [<tabletnode addr>]                                        \n\
                show all tabletnodes or single tabletnode info.             \n\
                (show more detail when using suffix \"x\")                  \n\
                                                                            \n\
       version\n\n";
}

void UsageMore(const std::string& prg_name) {
    std::cout << "\nSYNOPSIS\n";
    std::cout << "       " << prg_name << "  OPERATION  [OPTION...] \n\n";
    std::cout << "DESCRIPTION \n\
       tablet   <operation> <params>                                        \n\
           - operation                                                      \n\
                move    <tablet_path> <target_addr>                         \n\
                        move a tablet to target tabletnode                  \n\
                compact <tablet_path>                                       \n\
                split   <tablet_path>                                       \n\
                                                                            \n\
       safemode [get|enter|leave]                                           \n\
                                                                            \n\
       meta     [backup]                                                    \n\
                backup metatable in master memory                           \n\
                                                                            \n\
       meta2    [check|bak|show|repair]                                     \n\
                operate meta table.                                         \n\
                                                                            \n\
       findts   <tablename> <rowkey>                                        \n\
                find the specify tabletnode serving 'rowkey'.               \n\
                                                                            \n\
       version\n\n";
}
int32_t CreateOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 2) {
        Usage(argv[0]);
        return -1;
    }

    TableDescriptor table_desc;
    std::vector<std::string> delimiters;
    std::string schema = argv[2];
    if (!ParseSchema(schema, &table_desc)) {
        LOG(ERROR) << "fail to parse input table schema.";
        return -1;
    }
    if (argc == 4) {
        // have tablet delimiters
        std::string delim_file = argv[3];
        std::ifstream fin(delim_file.c_str());
        if (fin.fail()) {
            LOG(ERROR) << "fail to read delimiter file: " << delim_file;
            return -1;
        }
        std::string str;
        while (fin >> str) {
            delimiters.push_back(str);
        }
        bool is_delim_error = false;
        for (size_t i = 1; i < delimiters.size() - 1; i++) {
            if (delimiters[i] <= delimiters[i-1]) {
                LOG(ERROR) << "delimiter error: line: " << i + 1
                    << ", [" << delimiters[i] << "]";
                is_delim_error = true;
            }
        }
        if (is_delim_error) {
            LOG(ERROR) << "create table fail, delimiter error.";
            return -1;
        }
    } else if (argc > 4) {
        LOG(ERROR) << "too many args: " << argc;
        return -1;
    }
    if (!client->CreateTable(table_desc, delimiters, err)) {
        LOG(ERROR) << "fail to create table, "
            << strerr(*err);
        return -1;
    }
    ShowTableDescriptor(table_desc);
    return 0;
}

int32_t CreateByFileOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::ifstream fin(argv[2]);
    std::string schema;
    std::vector<std::string> delimiters;
    if (fin.good()) {
        std::string str;
        while (fin >> str) {
            schema.append(str);
        }
    }

    TableDescriptor table_desc;
    if (!ParseSchema(schema, &table_desc)) {
        LOG(ERROR) << "fail to parse input table schema.";
        return -1;
    }

    if (argc == 4) {
        std::ifstream fin(argv[3]);
        if (fin.fail()) {
            LOG(ERROR) << "fail to read delimiter file.";
            return -1;
        }
        std::string str;
        while (fin >> str) {
            delimiters.push_back(str);
        }
    }
    if (!client->CreateTable(table_desc, delimiters, err)) {
        LOG(ERROR) << "fail to create table, "
            << strerr(*err);
        return -1;
    }
    ShowTableDescriptor(table_desc);
    return 0;
}

int32_t UpdateOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 3) {
        Usage(argv[0]);
        return -1;
    }
    std::string schema = argv[2];
    PropTree schema_tree;
    if (!schema_tree.ParseFromString(schema)) {
        LOG(ERROR) << "[update] invalid schema: " << schema;
        LOG(ERROR) << "[update] state: " << schema_tree.State();
        return -1;
    }
    std::string tablename = schema_tree.GetRootNode()->name_;
    TableDescriptor* table_desc = client->GetTableDescriptor(tablename, err);
    if (table_desc == NULL) {
        LOG(ERROR) << "[update] can't get the TableDescriptor of table: " << tablename;
        return -1;
    }

    // if try to update lg or cf, need to disable table
    bool is_update_lg_cf = false;
    if (!UpdateTableDescriptor(schema_tree, table_desc, &is_update_lg_cf )) {
        LOG(ERROR) << "[update] update failed";
        return -1;
    }

    if (is_update_lg_cf && client->IsTableEnabled(table_desc->TableName(), err)) {
        LOG(ERROR) << "[update] table is enabled, disable it first: " << table_desc->TableName();
        return -1;
    }

    if (!client->UpdateTable(*table_desc, err)) {
        LOG(ERROR) << "[update] fail to update table, "
            << strerr(*err);
        return -1;
    }
    ShowTableDescriptor(*table_desc);
    delete table_desc;
    return 0;
}

int32_t DropOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    if (!client->DeleteTable(tablename, err)) {
        LOG(ERROR) << "fail to delete table";
        return -1;
    }
    return 0;
}

int32_t EnableOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    if (!client->EnableTable(tablename, err)) {
        LOG(ERROR) << "fail to enable table";
        return -1;
    }
    return 0;
}

int32_t DisableOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    if (!client->DisableTable(tablename, err)) {
        LOG(ERROR) << "fail to disable table";
        return -1;
    }
    return 0;
}

void ParseCfQualifier(const std::string& input, std::string* columnfamily, std::string* qualifier) {
    std::string::size_type pos = input.find(":", 0);
    if (pos != std::string::npos) {
        *columnfamily = input.substr(0, pos);
        *qualifier = input.substr(pos + 1);
    } else {
        *columnfamily = input;
        *qualifier = "";
    }
}

int32_t PutCounterOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    int64_t counter;
    if (!StringToNumber(value.c_str(), &counter)) {
       LOG(ERROR) << "invalid Integer number Got: " << value;
       return -1;
    }

    std::string s_counter = tera::CounterCoding::EncodeCounter(counter);
    if (!table->Put(rowkey, columnfamily, qualifier, s_counter, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t PutOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    if (!table->Put(rowkey, columnfamily, qualifier, value, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t PutTTLOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 6 && argc != 7) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    int32_t ttl = -1;
    if (argc == 6) {
        // use table as kv
        value = argv[4];
        ttl = atoi(argv[5]);
    } else if (argc == 7) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
        ttl = atoi(argv[6]);
    }
    if (!table->Put(rowkey, columnfamily, qualifier, value, ttl, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    return 0;
}

int32_t AppendOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    if (!table->Append(rowkey, columnfamily, qualifier, value, err)) {
        LOG(ERROR) << "fail to append record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t PutIfAbsentOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    if (!table->PutIfAbsent(rowkey, columnfamily, qualifier, value, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t AddOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR)<< "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    int64_t delta;
    if (!StringToNumber(value.c_str(), &delta)) {
        LOG(ERROR) << "invalid Integer number Got: " << value;
        return -1;
    }
    if (!table->Add(rowkey, columnfamily, qualifier, delta, err)) {
        LOG(ERROR) << "fail to add record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t GetOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 4 && argc != 5) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 4) {
        // use table as kv
    } else if (argc == 5) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
    }

    if (!table->Get(rowkey, columnfamily, qualifier, &value, err)) {
        LOG(ERROR) << "fail to get record from table: " << tablename;
        return -1;
    }

    std::cout << value;
    delete table;
    return 0;
}

int32_t GetCounterOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 4 && argc != 5) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 4) {
        // use table as kv
    } else if (argc == 5) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
    }

    if (!table->Get(rowkey, columnfamily, qualifier, &value, err)) {
        LOG(ERROR) << "fail to get record from table: " << tablename;
        return -1;
    }

    int64_t counter = 0;
    bool ret = tera::CounterCoding::DecodeCounter(value, &counter);
    if (!ret) {
        LOG(ERROR) << "invalid counter read, fail to parse";
    } else {
        std::cout << counter << std::endl;
    }
    delete table;
    return 0;
}


int32_t DeleteOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 4 && argc != 5) {
        LOG(ERROR) << "args number error: " << argc << ", need 4 | 5.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    std::string rowkey = argv[3];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string op = argv[1];
    RowMutation* mutation = table->NewRowMutation(rowkey);
    if (argc == 4) {
        // delete a row
        mutation->DeleteRow();
    } else if (argc == 5) {
        // delete a family or column
        std::string input(argv[4]);
        if (input.find(":", 0) == std::string::npos) {
            // delete a family
            mutation->DeleteFamily(input);
        } else {
            std::string family;
            std::string qualifier;
            ParseCfQualifier(input, &family, &qualifier);
            if (op == "delete") {
                // delete a column (all versions)
                mutation->DeleteColumns(family, qualifier);
            } else if (op == "delete1v") {
                // delete the newest version
                mutation->DeleteColumn(family, qualifier);
            }
        }
    } else {
        LOG(FATAL) << "should not run here.";
    }
    table->ApplyMutation(mutation);

    delete table;
    return 0;
}

int32_t ScanOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6 && argc != 7) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6 | 7.";
        Usage(argv[0]);
        return -1;
    }

    std::string op = argv[1];
    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string start_rowkey = argv[3];
    std::string end_rowkey = argv[4];
    ResultStream* result_stream;
    ScanDescriptor desc(start_rowkey);
    desc.SetEnd(end_rowkey);
    desc.SetBufferSize(FLAGS_tera_client_scan_package_size << 10);
    desc.SetAsync(FLAGS_tera_client_scan_async_enabled);

    if (argc == 5) {
        // scan all cells
    } else if (argc == 6) {
        if (!ParseScanSchema(argv[5], &desc)) {
            LOG(ERROR) << "fail to parse scan schema: " << argv[5];
            return -1;
        }
    }
    if (op == "scanallv") {
        desc.SetMaxVersions(std::numeric_limits<int>::max());
    }

    uint64_t snapshot = 0;
    if (argc == 7) {
        std::stringstream is;
        is << std::string(argv[6]);
        is >> snapshot;
    }

    desc.SetSnapshot(snapshot);
    if ((result_stream = table->Scan(desc, err)) == NULL) {
        LOG(ERROR) << "fail to scan records from table: " << tablename;
        return -1;
    }
    g_start_time = time(NULL);
    while (!result_stream->Done(err)) {
        int32_t len = result_stream->RowName().size()
            + result_stream->ColumnName().size()
            + sizeof(result_stream->Timestamp())
            + result_stream->Value().size();
        g_total_size += len;
        g_key_num ++;
        g_cur_batch_num ++;
        std::cout << result_stream->RowName() << ":"
           << result_stream->ColumnName() << ":"
           << result_stream->Timestamp() << ":"
           << result_stream->Value() << std::endl;
        result_stream->Next();
        if (g_cur_batch_num >= FLAGS_tera_client_batch_put_num) {
            int32_t time_cur=time(NULL);
            uint32_t time_used = time_cur - g_start_time;
            LOG(INFO) << "Scaning " << g_key_num << " keys " << g_key_num/(time_used?time_used:1)
                   << " keys/S " << g_total_size/1024.0/1024/(time_used?time_used:1) << " MB/S ";
            g_cur_batch_num = 0;
            g_last_time = time_cur;
        }
    }
    if (err->GetType() != ErrorCode::kOK) {
        LOG(ERROR) << "fail to finish scan: " << err->GetReason();
        return -1;
    }
    g_end_time = time(NULL);
    g_used_time = g_end_time - g_start_time;
    LOG(INFO) << "Scan done " << g_key_num << " keys " << g_key_num/(g_used_time?g_used_time:1)
            <<" keys/S " << g_total_size/1024.0/1024/(g_used_time?g_used_time:1) << " MB/S ";
    delete table;
    return 0;
}

int32_t ShowTabletList(const TabletMetaList& tablet_list, bool is_server_addr, bool is_x) {
    TPrinter printer;
    int cols;
    std::vector<string> row;
    if (is_x) {
        if (is_server_addr) {
            cols = 14;
            printer.Reset(cols);
            printer.AddRow(cols,
                           " ", "server_addr", "path", "status", "size",
                           "isbusy", "lread", "read", "rspeed", "write",
                           "wspeed", "scan", "sspeed", "startkey");
        } else {
            cols = 13;
            printer.Reset(cols);
            printer.AddRow(cols,
                           " ", "path", "status", "size", "isbusy",
                           "lread", "read", "rspeed", "write", "wspeed",
                           "scan", "sspeed", "startkey");
        }

        for (int32_t i = 0; i < tablet_list.meta_size(); ++i) {
            const TabletMeta& meta = tablet_list.meta(i);
            row.clear();
            row.push_back(NumberToString(i));
            if (is_server_addr) {
                row.push_back(meta.server_addr());
            }
            row.push_back(meta.path());
            row.push_back(StatusCodeToString(meta.status()));

            std::string size_str =
                utils::ConvertByteToString(meta.table_size()) +
                "[";
            for (int l = 0; l < meta.lg_size_size(); ++l) {
                size_str += utils::ConvertByteToString(meta.lg_size(l));
                if (l < meta.lg_size_size() - 1) {
                    size_str += " ";
                }
            }
            size_str += "]";
            row.push_back(size_str);

            if (tablet_list.counter_size() > 0) {
                const TabletCounter& counter = tablet_list.counter(i);
                if (counter.is_on_busy()) {
                    row.push_back("true");
                } else {
                    row.push_back("false");
                }
                row.push_back(NumberToString(counter.low_read_cell()));
                row.push_back(NumberToString(counter.read_rows()));
                row.push_back(utils::ConvertByteToString(counter.read_size()) + "B/s");
                row.push_back(NumberToString(counter.write_rows()));
                row.push_back(utils::ConvertByteToString(counter.write_size()) + "B/s");
                row.push_back(NumberToString(counter.scan_rows()));
                row.push_back(utils::ConvertByteToString(counter.scan_size()) + "B/s");
            }
            row.push_back(meta.key_range().key_start().substr(0, 20));
            printer.AddRow(row);
        }
    } else {
        cols = 7;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "server_addr", "path", "status",
                       "size", "startkey", "endkey");
        for (int32_t i = 0; i < tablet_list.meta_size(); ++i) {
            const TabletMeta& meta = tablet_list.meta(i);
            row.clear();
            row.push_back(NumberToString(i));
            row.push_back(meta.server_addr());
            row.push_back(meta.path());
            row.push_back(StatusCodeToString(meta.status()));
            row.push_back(utils::ConvertByteToString(meta.table_size()));
            row.push_back(DebugString(meta.key_range().key_start()).substr(0, 20));
            row.push_back(DebugString(meta.key_range().key_end()).substr(0, 20));
            printer.AddRow(row);
        }
    }
    printer.Print();
    return 0;
}

int32_t ShowAllTables(Client* client, bool is_x, bool show_all, ErrorCode* err) {
    TableMetaList table_list;
    TabletMetaList tablet_list;
    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTablesInfo(&table_list, &tablet_list, err)) {
        LOG(ERROR) << "fail to get meta data from tera.";
        return -1;
    }

    TPrinter printer;
    int cols;
    if (is_x) {
        cols = 18;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "tablename", "status", "size", "lg_size",
                       "tablet", "busy", "notready", "lread", "read",
                       "rmax", "rspeed", "write", "wmax", "wspeed",
                       "scan", "smax", "sspeed");
    } else {
        cols = 7;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "tablename", "status", "size", "lg_size",
                       "tablet", "busy");
    }
    for (int32_t table_no = 0; table_no < table_list.meta_size(); ++table_no) {
        std::string tablename = table_list.meta(table_no).table_name();
        TableStatus status = table_list.meta(table_no).status();
        int64_t size = 0;
        uint32_t tablet = 0;
        uint32_t busy = 0;
        uint32_t notready = 0;
        uint32_t lread = 0;
        uint32_t read = 0;
        uint32_t rmax = 0;
        uint64_t rspeed = 0;
        uint32_t write = 0;
        uint32_t wmax = 0;
        uint64_t wspeed = 0;
        uint32_t scan = 0;
        uint32_t smax = 0;
        uint64_t sspeed = 0;
        int64_t lg_num = 0;
        std::vector<int64_t> lg_size;
        for (int32_t i = 0; i < tablet_list.meta_size(); ++i) {
            if (tablet_list.meta(i).table_name() == tablename) {
                size += tablet_list.meta(i).table_size();
                tablet++;
                if (tablet_list.counter(i).is_on_busy()) {
                    busy++;
                }
                if (tablet_list.meta(i).status() != kTableReady) {
                    notready++;
                }
                lread += tablet_list.counter(i).low_read_cell();
                read += tablet_list.counter(i).read_rows();
                if (tablet_list.counter(i).read_rows() > rmax) {
                    rmax = tablet_list.counter(i).read_rows();
                }
                rspeed += tablet_list.counter(i).read_size();
                write += tablet_list.counter(i).write_rows();
                if (tablet_list.counter(i).write_rows() > wmax) {
                    wmax = tablet_list.counter(i).write_rows();
                }
                wspeed += tablet_list.counter(i).write_size();
                scan += tablet_list.counter(i).scan_rows();
                if (tablet_list.counter(i).scan_rows() > smax) {
                    smax = tablet_list.counter(i).scan_rows();
                }
                sspeed += tablet_list.counter(i).scan_size();

                if (lg_num == 0) {
                    lg_num = tablet_list.meta(i).lg_size_size();
                    lg_size.resize(lg_num, 0);
                }
                for (int l = 0; l < lg_num; ++l) {
                    lg_size[l] += tablet_list.meta(i).lg_size(l);
                }
            }
        }

        std::string lg_size_str = "";
        for (int l = 0; l < lg_num; ++l) {
            lg_size_str += utils::ConvertByteToString(lg_size[l]);
            if (l < lg_num - 1) {
                lg_size_str += " ";
            }
        }
        lg_size_str += "";
        if (is_x) {
            printer.AddRow(cols,
                           NumberToString(table_no).data(),
                           tablename.data(),
                           StatusCodeToString(status).data(),
                           utils::ConvertByteToString(size).data(),
                           lg_size_str.data(),
                           NumberToString(tablet).data(),
                           NumberToString(busy).data(),
                           NumberToString(notready).data(),
                           utils::ConvertByteToString(lread).data(),
                           utils::ConvertByteToString(read).data(),
                           utils::ConvertByteToString(rmax).data(),
                           (utils::ConvertByteToString(rspeed) + "B/s").data(),
                           utils::ConvertByteToString(write).data(),
                           utils::ConvertByteToString(wmax).data(),
                           (utils::ConvertByteToString(wspeed) + "B/s").data(),
                           utils::ConvertByteToString(scan).data(),
                           utils::ConvertByteToString(smax).data(),
                           (utils::ConvertByteToString(sspeed) + "B/s").data());
        } else {
            printer.AddRow(cols,
                           NumberToString(table_no).data(),
                           tablename.data(),
                           StatusCodeToString(status).data(),
                           utils::ConvertByteToString(size).data(),
                           lg_size_str.data(),
                           NumberToString(tablet).data(),
                           NumberToString(busy).data());
        }
    }
    printer.Print();
    std::cout << std::endl;
    if (show_all) {
        ShowTabletList(tablet_list, true, true);
    }
    return 0;
}

int32_t ShowSingleTable(Client* client, const string& table_name,
                        bool is_x, ErrorCode* err) {
    TableMeta table_meta;
    TabletMetaList tablet_list;

    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTablesInfo(table_name, &table_meta, &tablet_list, err)) {
        LOG(ERROR) << "table not exist: " << table_name;
        return -1;
    }

    std::cout << std::endl;
    ShowTabletList(tablet_list, true, is_x);
    std::cout << std::endl;
    return 0;
}

int32_t ShowSingleTabletNodeInfo(Client* client, const string& addr,
                                 bool is_x, ErrorCode* err) {
    TabletNodeInfo info;
    TabletMetaList tablet_list;
    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTabletNodesInfo(addr, &info, &tablet_list, err)) {
        LOG(ERROR) << "fail to show tabletnode: " << addr;
        return -1;
    }

    std::cout << "\nTabletNode Info:\n";
    std::cout << "  address:  " << info.addr() << std::endl;
    std::cout << "  status:   " << info.status_m() << "\n\n";
    int cols = 5;
    TPrinter printer(cols);
    printer.AddRow(cols, "workload", "tablets", "load", "busy", "split");
    std::vector<string> row;
    row.push_back(utils::ConvertByteToString(info.load()));
    row.push_back(NumberToString(info.tablet_total()));
    row.push_back(NumberToString(info.tablet_onload()));
    row.push_back(NumberToString(info.tablet_onbusy()));
    row.push_back(NumberToString(info.tablet_onsplit()));
    printer.AddRow(row);
    printer.Print();

    std::cout << std::endl;
    cols = 7;
    printer.Reset(cols);
    printer.AddRow(cols, "lread", "read", "rspeed", "write", "wspeed", "scan", "sspeed");
    row.clear();
    row.push_back(NumberToString(info.low_read_cell()));
    row.push_back(NumberToString(info.read_rows()));
    row.push_back(utils::ConvertByteToString(info.read_size()) + "B/s");
    row.push_back(NumberToString(info.write_rows()));
    row.push_back(utils::ConvertByteToString(info.write_size()) + "B/s");
    row.push_back(NumberToString(info.scan_rows()));
    row.push_back(utils::ConvertByteToString(info.scan_size()) + "B/s");
    printer.AddRow(row);
    printer.Print();

    std::cout << "\nHardware Info:\n";
    cols = 7;
    printer.Reset(cols);
    row.clear();
    printer.AddRow(cols, "mem_used", "net_tx", "net_rx", "dfs_r", "dfs_w", "local_r", "local_w");
    row.push_back(utils::ConvertByteToString(info.mem_used()));
    row.push_back(utils::ConvertByteToString(info.net_tx()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.net_rx()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.dfs_io_r()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.dfs_io_w()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.local_io_r()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.local_io_w()) + "B/s");
    printer.AddRow(row);
    printer.Print();

    std::cout << "\nOther Infos:\n";
    cols = info.extra_info_size();
    printer.Reset(cols);
    row.clear();
    for (int i = 0; i < cols; ++i) {
        row.push_back(info.extra_info(i).name());
    }
    printer.AddRow(row);
    std::vector<int64_t> row_int;
    for (int i = 0; i < cols; ++i) {
        row_int.push_back(info.extra_info(i).value());
    }
    printer.AddRow(row_int);
    printer.Print();

    std::cout << "\nTablets In this TabletNode:\n";
    ShowTabletList(tablet_list, false, is_x);
    return 0;
}

int32_t ShowTabletNodesInfo(Client* client, bool is_x, ErrorCode* err) {
    std::vector<TabletNodeInfo> infos;
    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTabletNodesInfo(&infos, err)) {
        LOG(ERROR) << "fail to get meta data from tera.";
        return -1;
    }

    int cols;
    TPrinter printer;
    if (is_x) {
        cols = 23;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "address", "status", "size", "num",
                       "lread", "r", "rspd", "w", "wspd",
                       "s", "sspd", "rdly", "rp", "sp",
                       "wp", "ld", "bs", "mem", "net_tx",
                       "net_rx", "dfs_r", "dfs_w");
        std::vector<string> row;
        for (size_t i = 0; i < infos.size(); ++i) {
            std::map<string, string> extra;
            for (int j = 0; j < infos[i].extra_info_size(); ++j) {
                extra[infos[i].extra_info(j).name()] =
                    NumberToString(infos[i].extra_info(j).value());
            }

            row.clear();
            row.push_back(NumberToString(i));
            row.push_back(infos[i].addr());
            row.push_back(infos[i].status_m());
            row.push_back(utils::ConvertByteToString(infos[i].load()));
            row.push_back(NumberToString(infos[i].tablet_total()));
            row.push_back(NumberToString(infos[i].low_read_cell()));
            row.push_back(NumberToString(infos[i].read_rows()));
            row.push_back(utils::ConvertByteToString(infos[i].read_size()) + "B");
            row.push_back(NumberToString(infos[i].write_rows()));
            row.push_back(utils::ConvertByteToString(infos[i].write_size()) + "B");
            row.push_back(NumberToString(infos[i].scan_rows()));
            row.push_back(utils::ConvertByteToString(infos[i].scan_size()) + "B");
            row.push_back(extra["rand_read_delay"] + "ms");
            row.push_back(extra["read_pending"]);
            row.push_back(extra["write_pending"]);
            row.push_back(extra["scan_pending"]);
            row.push_back(NumberToString(infos[i].tablet_onload()));
            row.push_back(NumberToString(infos[i].tablet_onbusy()));
            row.push_back(utils::ConvertByteToString(infos[i].mem_used()));
            row.push_back(utils::ConvertByteToString(infos[i].net_tx()));
            row.push_back(utils::ConvertByteToString(infos[i].net_rx()));
            row.push_back(utils::ConvertByteToString(infos[i].dfs_io_r()));
            row.push_back(utils::ConvertByteToString(infos[i].dfs_io_w()));
            printer.AddRow(row);
        }
    } else {
        cols = 7;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "address", "status", "workload",
                       "tablet", "load", "busy");
        std::vector<string> row;
        for (size_t i = 0; i < infos.size(); ++i) {
            row.clear();
            row.push_back(NumberToString(i));
            row.push_back(infos[i].addr());
            row.push_back(infos[i].status_m());
            row.push_back(utils::ConvertByteToString(infos[i].load()));
            row.push_back(NumberToString(infos[i].tablet_total()));
            row.push_back(NumberToString(infos[i].tablet_onload()));
            row.push_back(NumberToString(infos[i].tablet_onbusy()));
            printer.AddRow(row);
        }<commit_msg>show cpu info in 'teracli showtsx'<commit_after>// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//

#include <stdlib.h>
#include <unistd.h>

#include <fstream>
#include <iostream>
#include <limits>
#include <sstream>

#include <gflags/gflags.h>
#include <glog/logging.h>

#include "common/base/string_ext.h"
#include "common/base/string_number.h"
#include "common/file/file_path.h"
#include "io/coding.h"
#include "proto/kv_helper.h"
#include "proto/proto_helper.h"
#include "proto/tabletnode.pb.h"
#include "proto/tabletnode_client.h"
#include "sdk/client_impl.h"
#include "sdk/sdk_utils.h"
#include "sdk/sdk_zk.h"
#include "sdk/table_impl.h"
#include "sdk/tera.h"
#include "utils/string_util.h"
#include "utils/tprinter.h"
#include "utils/utils_cmd.h"
#include "version.h"

DECLARE_string(flagfile);
DECLARE_string(log_dir);
DECLARE_string(tera_master_meta_table_name);
DECLARE_string(tera_zk_addr_list);
DECLARE_string(tera_zk_root_path);

DEFINE_int32(tera_client_batch_put_num, 1000, "num of each batch in batch put mode");
DEFINE_int32(tera_client_scan_package_size, 1024, "the package size (in KB) of each scan request");
DEFINE_bool(tera_client_scan_async_enabled, false, "enable the streaming scan mode");

volatile int32_t g_start_time = 0;
volatile int32_t g_end_time = 0;
volatile int32_t g_used_time = 0;
volatile int32_t g_last_time  = 0;
volatile int64_t g_total_size = 0;
volatile int32_t g_key_num = 0;
Mutex g_stat_lock;

volatile int32_t g_cur_batch_num = 0;

using namespace tera;

void Usage(const std::string& prg_name) {
    std::cout << "\nSYNOPSIS\n";
    std::cout << "       " << prg_name << "  OPERATION  [OPTION...] \n\n";
    std::cout << "DESCRIPTION \n\
       create   <schema> [<delimiter_file>]                                 \n\
              - schema syntax (all properties are optional):                \n\
                    tablename <rawkey=binary,splitsize=1024,...> {          \n\
                        lg1 <storage=flash,...> {                           \n\
                            cf1 <maxversion=3>,                             \n\
                            cf2...},                                        \n\
                        lg2...                                              \n\
                    }                                                       \n\
              - kv mode schema:                                             \n\
                    tablename <splitsize=1024, storage=memory, ...>         \n\
              - simple mode schema:                                         \n\
                    tablename{cf1, cf2, cf3, ...}                           \n\
                                                                            \n\
       createbyfile   <schema_file> [<delimiter_file>]                      \n\
                                                                            \n\
       update <schema>                                                      \n\
              - kv schema:                                                  \n\
                    e.g. tablename<splitsize=512,storage=disk>              \n\
              - table schema:                                               \n\
                - update properties                                         \n\
                    e.g. tablename<splitsize=512>                           \n\
                    e.g. tablename{lg0{cf0<ttl=123>}}                       \n\
                    e.g. tablename<mergesize=233>{lg0<storage=disk>{cf0<ttl=123>}}       \n\
                - add new cf                                                \n\
                    e.g. tablename{lg0{cf0<ttl=250>,new_cf<op=add,ttl=233>}}\n\
                - delete cf                                                 \n\
                    e.g. tablename{lg0{cf0<op=del>}}                        \n\
                                                                            \n\
       enable/disable/drop  <tablename>                                     \n\
                                                                            \n\
       put      <tablename> <rowkey> [<columnfamily:qualifier>] <value>     \n\
                                                                            \n\
       put-ttl  <tablename> <rowkey> [<columnfamily:qualifier>] <value> <ttl(second)>    \n\
                                                                            \n\
       putif    <tablename> <rowkey> [<columnfamily:qualifier>] <value>     \n\
                                                                            \n\
       get      <tablename> <rowkey> [<columnfamily:qualifier>]             \n\
                                                                            \n\
       scan[allv] <tablename> <startkey> <endkey> [<\"cf1|cf2\">]           \n\
                scan table from startkey to endkey.                         \n\
                (return all qulifier version when using suffix \"allv\")    \n\
                                                                            \n\
       delete[1v] <tablename> <rowkey> [<columnfamily:qualifier>]           \n\
                delete row/columnfamily/qualifiers.                         \n\
                (only delete latest version when using suffix \"1v\")       \n\
                                                                            \n\
       put_counter <tablename> <rowkey> [<columnfamily:qualifier>] <integer>\n\
                                                                            \n\
       get_counter <tablename> <rowkey> [<columnfamily:qualifier>]          \n\
                                                                            \n\
       add      <tablename> <rowkey> <columnfamily:qualifier>   delta       \n\
                add 'delta'(int64_t) to specified cell                      \n\
       append   <tablename> <rowkey> [<columnfamily:qualifier>] <value>     \n\
                                                                            \n\
       batchput <tablename> <input file>                                    \n\
                                                                            \n\
       batchget <tablename> <input file>                                    \n\
                                                                            \n\
       show[x]  [<tablename>]                                               \n\
                show table list or tablets info.                            \n\
                (show more detail when using suffix \"x\")                  \n\
                                                                            \n\
       showschema[x] <tablename>                                            \n\
                show table schema (show more detail when using suffix \"x\")\n\
                                                                            \n\
       showts[x] [<tabletnode addr>]                                        \n\
                show all tabletnodes or single tabletnode info.             \n\
                (show more detail when using suffix \"x\")                  \n\
                                                                            \n\
       version\n\n";
}

void UsageMore(const std::string& prg_name) {
    std::cout << "\nSYNOPSIS\n";
    std::cout << "       " << prg_name << "  OPERATION  [OPTION...] \n\n";
    std::cout << "DESCRIPTION \n\
       tablet   <operation> <params>                                        \n\
           - operation                                                      \n\
                move    <tablet_path> <target_addr>                         \n\
                        move a tablet to target tabletnode                  \n\
                compact <tablet_path>                                       \n\
                split   <tablet_path>                                       \n\
                                                                            \n\
       safemode [get|enter|leave]                                           \n\
                                                                            \n\
       meta     [backup]                                                    \n\
                backup metatable in master memory                           \n\
                                                                            \n\
       meta2    [check|bak|show|repair]                                     \n\
                operate meta table.                                         \n\
                                                                            \n\
       findts   <tablename> <rowkey>                                        \n\
                find the specify tabletnode serving 'rowkey'.               \n\
                                                                            \n\
       version\n\n";
}
int32_t CreateOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 2) {
        Usage(argv[0]);
        return -1;
    }

    TableDescriptor table_desc;
    std::vector<std::string> delimiters;
    std::string schema = argv[2];
    if (!ParseSchema(schema, &table_desc)) {
        LOG(ERROR) << "fail to parse input table schema.";
        return -1;
    }
    if (argc == 4) {
        // have tablet delimiters
        std::string delim_file = argv[3];
        std::ifstream fin(delim_file.c_str());
        if (fin.fail()) {
            LOG(ERROR) << "fail to read delimiter file: " << delim_file;
            return -1;
        }
        std::string str;
        while (fin >> str) {
            delimiters.push_back(str);
        }
        bool is_delim_error = false;
        for (size_t i = 1; i < delimiters.size() - 1; i++) {
            if (delimiters[i] <= delimiters[i-1]) {
                LOG(ERROR) << "delimiter error: line: " << i + 1
                    << ", [" << delimiters[i] << "]";
                is_delim_error = true;
            }
        }
        if (is_delim_error) {
            LOG(ERROR) << "create table fail, delimiter error.";
            return -1;
        }
    } else if (argc > 4) {
        LOG(ERROR) << "too many args: " << argc;
        return -1;
    }
    if (!client->CreateTable(table_desc, delimiters, err)) {
        LOG(ERROR) << "fail to create table, "
            << strerr(*err);
        return -1;
    }
    ShowTableDescriptor(table_desc);
    return 0;
}

int32_t CreateByFileOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::ifstream fin(argv[2]);
    std::string schema;
    std::vector<std::string> delimiters;
    if (fin.good()) {
        std::string str;
        while (fin >> str) {
            schema.append(str);
        }
    }

    TableDescriptor table_desc;
    if (!ParseSchema(schema, &table_desc)) {
        LOG(ERROR) << "fail to parse input table schema.";
        return -1;
    }

    if (argc == 4) {
        std::ifstream fin(argv[3]);
        if (fin.fail()) {
            LOG(ERROR) << "fail to read delimiter file.";
            return -1;
        }
        std::string str;
        while (fin >> str) {
            delimiters.push_back(str);
        }
    }
    if (!client->CreateTable(table_desc, delimiters, err)) {
        LOG(ERROR) << "fail to create table, "
            << strerr(*err);
        return -1;
    }
    ShowTableDescriptor(table_desc);
    return 0;
}

int32_t UpdateOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 3) {
        Usage(argv[0]);
        return -1;
    }
    std::string schema = argv[2];
    PropTree schema_tree;
    if (!schema_tree.ParseFromString(schema)) {
        LOG(ERROR) << "[update] invalid schema: " << schema;
        LOG(ERROR) << "[update] state: " << schema_tree.State();
        return -1;
    }
    std::string tablename = schema_tree.GetRootNode()->name_;
    TableDescriptor* table_desc = client->GetTableDescriptor(tablename, err);
    if (table_desc == NULL) {
        LOG(ERROR) << "[update] can't get the TableDescriptor of table: " << tablename;
        return -1;
    }

    // if try to update lg or cf, need to disable table
    bool is_update_lg_cf = false;
    if (!UpdateTableDescriptor(schema_tree, table_desc, &is_update_lg_cf )) {
        LOG(ERROR) << "[update] update failed";
        return -1;
    }

    if (is_update_lg_cf && client->IsTableEnabled(table_desc->TableName(), err)) {
        LOG(ERROR) << "[update] table is enabled, disable it first: " << table_desc->TableName();
        return -1;
    }

    if (!client->UpdateTable(*table_desc, err)) {
        LOG(ERROR) << "[update] fail to update table, "
            << strerr(*err);
        return -1;
    }
    ShowTableDescriptor(*table_desc);
    delete table_desc;
    return 0;
}

int32_t DropOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    if (!client->DeleteTable(tablename, err)) {
        LOG(ERROR) << "fail to delete table";
        return -1;
    }
    return 0;
}

int32_t EnableOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    if (!client->EnableTable(tablename, err)) {
        LOG(ERROR) << "fail to enable table";
        return -1;
    }
    return 0;
}

int32_t DisableOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc < 3) {
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    if (!client->DisableTable(tablename, err)) {
        LOG(ERROR) << "fail to disable table";
        return -1;
    }
    return 0;
}

void ParseCfQualifier(const std::string& input, std::string* columnfamily, std::string* qualifier) {
    std::string::size_type pos = input.find(":", 0);
    if (pos != std::string::npos) {
        *columnfamily = input.substr(0, pos);
        *qualifier = input.substr(pos + 1);
    } else {
        *columnfamily = input;
        *qualifier = "";
    }
}

int32_t PutCounterOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    int64_t counter;
    if (!StringToNumber(value.c_str(), &counter)) {
       LOG(ERROR) << "invalid Integer number Got: " << value;
       return -1;
    }

    std::string s_counter = tera::CounterCoding::EncodeCounter(counter);
    if (!table->Put(rowkey, columnfamily, qualifier, s_counter, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t PutOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    if (!table->Put(rowkey, columnfamily, qualifier, value, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t PutTTLOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 6 && argc != 7) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    int32_t ttl = -1;
    if (argc == 6) {
        // use table as kv
        value = argv[4];
        ttl = atoi(argv[5]);
    } else if (argc == 7) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
        ttl = atoi(argv[6]);
    }
    if (!table->Put(rowkey, columnfamily, qualifier, value, ttl, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    return 0;
}

int32_t AppendOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    if (!table->Append(rowkey, columnfamily, qualifier, value, err)) {
        LOG(ERROR) << "fail to append record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t PutIfAbsentOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    if (!table->PutIfAbsent(rowkey, columnfamily, qualifier, value, err)) {
        LOG(ERROR) << "fail to put record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t AddOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6) {
        LOG(ERROR)<< "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 5) {
        // use table as kv
        value = argv[4];
    } else if (argc == 6) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
        value = argv[5];
    }
    int64_t delta;
    if (!StringToNumber(value.c_str(), &delta)) {
        LOG(ERROR) << "invalid Integer number Got: " << value;
        return -1;
    }
    if (!table->Add(rowkey, columnfamily, qualifier, delta, err)) {
        LOG(ERROR) << "fail to add record to table: " << tablename;
        return -1;
    }
    delete table;
    return 0;
}

int32_t GetOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 4 && argc != 5) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 4) {
        // use table as kv
    } else if (argc == 5) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
    }

    if (!table->Get(rowkey, columnfamily, qualifier, &value, err)) {
        LOG(ERROR) << "fail to get record from table: " << tablename;
        return -1;
    }

    std::cout << value;
    delete table;
    return 0;
}

int32_t GetCounterOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 4 && argc != 5) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string rowkey = argv[3];
    std::string columnfamily = "";
    std::string qualifier = "";
    std::string value;
    if (argc == 4) {
        // use table as kv
    } else if (argc == 5) {
        ParseCfQualifier(argv[4], &columnfamily, &qualifier);
    }

    if (!table->Get(rowkey, columnfamily, qualifier, &value, err)) {
        LOG(ERROR) << "fail to get record from table: " << tablename;
        return -1;
    }

    int64_t counter = 0;
    bool ret = tera::CounterCoding::DecodeCounter(value, &counter);
    if (!ret) {
        LOG(ERROR) << "invalid counter read, fail to parse";
    } else {
        std::cout << counter << std::endl;
    }
    delete table;
    return 0;
}


int32_t DeleteOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 4 && argc != 5) {
        LOG(ERROR) << "args number error: " << argc << ", need 4 | 5.";
        Usage(argv[0]);
        return -1;
    }

    std::string tablename = argv[2];
    std::string rowkey = argv[3];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string op = argv[1];
    RowMutation* mutation = table->NewRowMutation(rowkey);
    if (argc == 4) {
        // delete a row
        mutation->DeleteRow();
    } else if (argc == 5) {
        // delete a family or column
        std::string input(argv[4]);
        if (input.find(":", 0) == std::string::npos) {
            // delete a family
            mutation->DeleteFamily(input);
        } else {
            std::string family;
            std::string qualifier;
            ParseCfQualifier(input, &family, &qualifier);
            if (op == "delete") {
                // delete a column (all versions)
                mutation->DeleteColumns(family, qualifier);
            } else if (op == "delete1v") {
                // delete the newest version
                mutation->DeleteColumn(family, qualifier);
            }
        }
    } else {
        LOG(FATAL) << "should not run here.";
    }
    table->ApplyMutation(mutation);

    delete table;
    return 0;
}

int32_t ScanOp(Client* client, int32_t argc, char** argv, ErrorCode* err) {
    if (argc != 5 && argc != 6 && argc != 7) {
        LOG(ERROR) << "args number error: " << argc << ", need 5 | 6 | 7.";
        Usage(argv[0]);
        return -1;
    }

    std::string op = argv[1];
    std::string tablename = argv[2];
    Table* table = NULL;
    if ((table = client->OpenTable(tablename, err)) == NULL) {
        LOG(ERROR) << "fail to open table";
        return -1;
    }

    std::string start_rowkey = argv[3];
    std::string end_rowkey = argv[4];
    ResultStream* result_stream;
    ScanDescriptor desc(start_rowkey);
    desc.SetEnd(end_rowkey);
    desc.SetBufferSize(FLAGS_tera_client_scan_package_size << 10);
    desc.SetAsync(FLAGS_tera_client_scan_async_enabled);

    if (argc == 5) {
        // scan all cells
    } else if (argc == 6) {
        if (!ParseScanSchema(argv[5], &desc)) {
            LOG(ERROR) << "fail to parse scan schema: " << argv[5];
            return -1;
        }
    }
    if (op == "scanallv") {
        desc.SetMaxVersions(std::numeric_limits<int>::max());
    }

    uint64_t snapshot = 0;
    if (argc == 7) {
        std::stringstream is;
        is << std::string(argv[6]);
        is >> snapshot;
    }

    desc.SetSnapshot(snapshot);
    if ((result_stream = table->Scan(desc, err)) == NULL) {
        LOG(ERROR) << "fail to scan records from table: " << tablename;
        return -1;
    }
    g_start_time = time(NULL);
    while (!result_stream->Done(err)) {
        int32_t len = result_stream->RowName().size()
            + result_stream->ColumnName().size()
            + sizeof(result_stream->Timestamp())
            + result_stream->Value().size();
        g_total_size += len;
        g_key_num ++;
        g_cur_batch_num ++;
        std::cout << result_stream->RowName() << ":"
           << result_stream->ColumnName() << ":"
           << result_stream->Timestamp() << ":"
           << result_stream->Value() << std::endl;
        result_stream->Next();
        if (g_cur_batch_num >= FLAGS_tera_client_batch_put_num) {
            int32_t time_cur=time(NULL);
            uint32_t time_used = time_cur - g_start_time;
            LOG(INFO) << "Scaning " << g_key_num << " keys " << g_key_num/(time_used?time_used:1)
                   << " keys/S " << g_total_size/1024.0/1024/(time_used?time_used:1) << " MB/S ";
            g_cur_batch_num = 0;
            g_last_time = time_cur;
        }
    }
    if (err->GetType() != ErrorCode::kOK) {
        LOG(ERROR) << "fail to finish scan: " << err->GetReason();
        return -1;
    }
    g_end_time = time(NULL);
    g_used_time = g_end_time - g_start_time;
    LOG(INFO) << "Scan done " << g_key_num << " keys " << g_key_num/(g_used_time?g_used_time:1)
            <<" keys/S " << g_total_size/1024.0/1024/(g_used_time?g_used_time:1) << " MB/S ";
    delete table;
    return 0;
}

int32_t ShowTabletList(const TabletMetaList& tablet_list, bool is_server_addr, bool is_x) {
    TPrinter printer;
    int cols;
    std::vector<string> row;
    if (is_x) {
        if (is_server_addr) {
            cols = 14;
            printer.Reset(cols);
            printer.AddRow(cols,
                           " ", "server_addr", "path", "status", "size",
                           "isbusy", "lread", "read", "rspeed", "write",
                           "wspeed", "scan", "sspeed", "startkey");
        } else {
            cols = 13;
            printer.Reset(cols);
            printer.AddRow(cols,
                           " ", "path", "status", "size", "isbusy",
                           "lread", "read", "rspeed", "write", "wspeed",
                           "scan", "sspeed", "startkey");
        }

        for (int32_t i = 0; i < tablet_list.meta_size(); ++i) {
            const TabletMeta& meta = tablet_list.meta(i);
            row.clear();
            row.push_back(NumberToString(i));
            if (is_server_addr) {
                row.push_back(meta.server_addr());
            }
            row.push_back(meta.path());
            row.push_back(StatusCodeToString(meta.status()));

            std::string size_str =
                utils::ConvertByteToString(meta.table_size()) +
                "[";
            for (int l = 0; l < meta.lg_size_size(); ++l) {
                size_str += utils::ConvertByteToString(meta.lg_size(l));
                if (l < meta.lg_size_size() - 1) {
                    size_str += " ";
                }
            }
            size_str += "]";
            row.push_back(size_str);

            if (tablet_list.counter_size() > 0) {
                const TabletCounter& counter = tablet_list.counter(i);
                if (counter.is_on_busy()) {
                    row.push_back("true");
                } else {
                    row.push_back("false");
                }
                row.push_back(NumberToString(counter.low_read_cell()));
                row.push_back(NumberToString(counter.read_rows()));
                row.push_back(utils::ConvertByteToString(counter.read_size()) + "B/s");
                row.push_back(NumberToString(counter.write_rows()));
                row.push_back(utils::ConvertByteToString(counter.write_size()) + "B/s");
                row.push_back(NumberToString(counter.scan_rows()));
                row.push_back(utils::ConvertByteToString(counter.scan_size()) + "B/s");
            }
            row.push_back(meta.key_range().key_start().substr(0, 20));
            printer.AddRow(row);
        }
    } else {
        cols = 7;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "server_addr", "path", "status",
                       "size", "startkey", "endkey");
        for (int32_t i = 0; i < tablet_list.meta_size(); ++i) {
            const TabletMeta& meta = tablet_list.meta(i);
            row.clear();
            row.push_back(NumberToString(i));
            row.push_back(meta.server_addr());
            row.push_back(meta.path());
            row.push_back(StatusCodeToString(meta.status()));
            row.push_back(utils::ConvertByteToString(meta.table_size()));
            row.push_back(DebugString(meta.key_range().key_start()).substr(0, 20));
            row.push_back(DebugString(meta.key_range().key_end()).substr(0, 20));
            printer.AddRow(row);
        }
    }
    printer.Print();
    return 0;
}

int32_t ShowAllTables(Client* client, bool is_x, bool show_all, ErrorCode* err) {
    TableMetaList table_list;
    TabletMetaList tablet_list;
    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTablesInfo(&table_list, &tablet_list, err)) {
        LOG(ERROR) << "fail to get meta data from tera.";
        return -1;
    }

    TPrinter printer;
    int cols;
    if (is_x) {
        cols = 18;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "tablename", "status", "size", "lg_size",
                       "tablet", "busy", "notready", "lread", "read",
                       "rmax", "rspeed", "write", "wmax", "wspeed",
                       "scan", "smax", "sspeed");
    } else {
        cols = 7;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "tablename", "status", "size", "lg_size",
                       "tablet", "busy");
    }
    for (int32_t table_no = 0; table_no < table_list.meta_size(); ++table_no) {
        std::string tablename = table_list.meta(table_no).table_name();
        TableStatus status = table_list.meta(table_no).status();
        int64_t size = 0;
        uint32_t tablet = 0;
        uint32_t busy = 0;
        uint32_t notready = 0;
        uint32_t lread = 0;
        uint32_t read = 0;
        uint32_t rmax = 0;
        uint64_t rspeed = 0;
        uint32_t write = 0;
        uint32_t wmax = 0;
        uint64_t wspeed = 0;
        uint32_t scan = 0;
        uint32_t smax = 0;
        uint64_t sspeed = 0;
        int64_t lg_num = 0;
        std::vector<int64_t> lg_size;
        for (int32_t i = 0; i < tablet_list.meta_size(); ++i) {
            if (tablet_list.meta(i).table_name() == tablename) {
                size += tablet_list.meta(i).table_size();
                tablet++;
                if (tablet_list.counter(i).is_on_busy()) {
                    busy++;
                }
                if (tablet_list.meta(i).status() != kTableReady) {
                    notready++;
                }
                lread += tablet_list.counter(i).low_read_cell();
                read += tablet_list.counter(i).read_rows();
                if (tablet_list.counter(i).read_rows() > rmax) {
                    rmax = tablet_list.counter(i).read_rows();
                }
                rspeed += tablet_list.counter(i).read_size();
                write += tablet_list.counter(i).write_rows();
                if (tablet_list.counter(i).write_rows() > wmax) {
                    wmax = tablet_list.counter(i).write_rows();
                }
                wspeed += tablet_list.counter(i).write_size();
                scan += tablet_list.counter(i).scan_rows();
                if (tablet_list.counter(i).scan_rows() > smax) {
                    smax = tablet_list.counter(i).scan_rows();
                }
                sspeed += tablet_list.counter(i).scan_size();

                if (lg_num == 0) {
                    lg_num = tablet_list.meta(i).lg_size_size();
                    lg_size.resize(lg_num, 0);
                }
                for (int l = 0; l < lg_num; ++l) {
                    lg_size[l] += tablet_list.meta(i).lg_size(l);
                }
            }
        }

        std::string lg_size_str = "";
        for (int l = 0; l < lg_num; ++l) {
            lg_size_str += utils::ConvertByteToString(lg_size[l]);
            if (l < lg_num - 1) {
                lg_size_str += " ";
            }
        }
        lg_size_str += "";
        if (is_x) {
            printer.AddRow(cols,
                           NumberToString(table_no).data(),
                           tablename.data(),
                           StatusCodeToString(status).data(),
                           utils::ConvertByteToString(size).data(),
                           lg_size_str.data(),
                           NumberToString(tablet).data(),
                           NumberToString(busy).data(),
                           NumberToString(notready).data(),
                           utils::ConvertByteToString(lread).data(),
                           utils::ConvertByteToString(read).data(),
                           utils::ConvertByteToString(rmax).data(),
                           (utils::ConvertByteToString(rspeed) + "B/s").data(),
                           utils::ConvertByteToString(write).data(),
                           utils::ConvertByteToString(wmax).data(),
                           (utils::ConvertByteToString(wspeed) + "B/s").data(),
                           utils::ConvertByteToString(scan).data(),
                           utils::ConvertByteToString(smax).data(),
                           (utils::ConvertByteToString(sspeed) + "B/s").data());
        } else {
            printer.AddRow(cols,
                           NumberToString(table_no).data(),
                           tablename.data(),
                           StatusCodeToString(status).data(),
                           utils::ConvertByteToString(size).data(),
                           lg_size_str.data(),
                           NumberToString(tablet).data(),
                           NumberToString(busy).data());
        }
    }
    printer.Print();
    std::cout << std::endl;
    if (show_all) {
        ShowTabletList(tablet_list, true, true);
    }
    return 0;
}

int32_t ShowSingleTable(Client* client, const string& table_name,
                        bool is_x, ErrorCode* err) {
    TableMeta table_meta;
    TabletMetaList tablet_list;

    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTablesInfo(table_name, &table_meta, &tablet_list, err)) {
        LOG(ERROR) << "table not exist: " << table_name;
        return -1;
    }

    std::cout << std::endl;
    ShowTabletList(tablet_list, true, is_x);
    std::cout << std::endl;
    return 0;
}

int32_t ShowSingleTabletNodeInfo(Client* client, const string& addr,
                                 bool is_x, ErrorCode* err) {
    TabletNodeInfo info;
    TabletMetaList tablet_list;
    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTabletNodesInfo(addr, &info, &tablet_list, err)) {
        LOG(ERROR) << "fail to show tabletnode: " << addr;
        return -1;
    }

    std::cout << "\nTabletNode Info:\n";
    std::cout << "  address:  " << info.addr() << std::endl;
    std::cout << "  status:   " << info.status_m() << "\n\n";
    int cols = 5;
    TPrinter printer(cols);
    printer.AddRow(cols, "workload", "tablets", "load", "busy", "split");
    std::vector<string> row;
    row.push_back(utils::ConvertByteToString(info.load()));
    row.push_back(NumberToString(info.tablet_total()));
    row.push_back(NumberToString(info.tablet_onload()));
    row.push_back(NumberToString(info.tablet_onbusy()));
    row.push_back(NumberToString(info.tablet_onsplit()));
    printer.AddRow(row);
    printer.Print();

    std::cout << std::endl;
    cols = 7;
    printer.Reset(cols);
    printer.AddRow(cols, "lread", "read", "rspeed", "write", "wspeed", "scan", "sspeed");
    row.clear();
    row.push_back(NumberToString(info.low_read_cell()));
    row.push_back(NumberToString(info.read_rows()));
    row.push_back(utils::ConvertByteToString(info.read_size()) + "B/s");
    row.push_back(NumberToString(info.write_rows()));
    row.push_back(utils::ConvertByteToString(info.write_size()) + "B/s");
    row.push_back(NumberToString(info.scan_rows()));
    row.push_back(utils::ConvertByteToString(info.scan_size()) + "B/s");
    printer.AddRow(row);
    printer.Print();

    std::cout << "\nHardware Info:\n";
    cols = 7;
    printer.Reset(cols);
    row.clear();
    printer.AddRow(cols, "mem_used", "net_tx", "net_rx", "dfs_r", "dfs_w", "local_r", "local_w");
    row.push_back(utils::ConvertByteToString(info.mem_used()));
    row.push_back(utils::ConvertByteToString(info.net_tx()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.net_rx()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.dfs_io_r()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.dfs_io_w()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.local_io_r()) + "B/s");
    row.push_back(utils::ConvertByteToString(info.local_io_w()) + "B/s");
    printer.AddRow(row);
    printer.Print();

    std::cout << "\nOther Infos:\n";
    cols = info.extra_info_size();
    printer.Reset(cols);
    row.clear();
    for (int i = 0; i < cols; ++i) {
        row.push_back(info.extra_info(i).name());
    }
    printer.AddRow(row);
    std::vector<int64_t> row_int;
    for (int i = 0; i < cols; ++i) {
        row_int.push_back(info.extra_info(i).value());
    }
    printer.AddRow(row_int);
    printer.Print();

    std::cout << "\nTablets In this TabletNode:\n";
    ShowTabletList(tablet_list, false, is_x);
    return 0;
}

int32_t ShowTabletNodesInfo(Client* client, bool is_x, ErrorCode* err) {
    std::vector<TabletNodeInfo> infos;
    tera::ClientImpl* client_impl = static_cast<tera::ClientImpl*>(client);
    if (!client_impl->ShowTabletNodesInfo(&infos, err)) {
        LOG(ERROR) << "fail to get meta data from tera.";
        return -1;
    }

    int cols;
    TPrinter printer;
    if (is_x) {
        cols = 24;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "address", "status", "size", "num",
                       "lread", "r", "rspd", "w", "wspd",
                       "s", "sspd", "rdly", "rp", "sp",
                       "wp", "ld", "bs", "mem", "cpu",
                       "net_tx", "net_rx", "dfs_r", "dfs_w");
        std::vector<string> row;
        for (size_t i = 0; i < infos.size(); ++i) {
            std::map<string, string> extra;
            for (int j = 0; j < infos[i].extra_info_size(); ++j) {
                extra[infos[i].extra_info(j).name()] =
                    NumberToString(infos[i].extra_info(j).value());
            }

            row.clear();
            row.push_back(NumberToString(i));
            row.push_back(infos[i].addr());
            row.push_back(infos[i].status_m());
            row.push_back(utils::ConvertByteToString(infos[i].load()));
            row.push_back(NumberToString(infos[i].tablet_total()));
            row.push_back(NumberToString(infos[i].low_read_cell()));
            row.push_back(NumberToString(infos[i].read_rows()));
            row.push_back(utils::ConvertByteToString(infos[i].read_size()) + "B");
            row.push_back(NumberToString(infos[i].write_rows()));
            row.push_back(utils::ConvertByteToString(infos[i].write_size()) + "B");
            row.push_back(NumberToString(infos[i].scan_rows()));
            row.push_back(utils::ConvertByteToString(infos[i].scan_size()) + "B");
            row.push_back(extra["rand_read_delay"] + "ms");
            row.push_back(extra["read_pending"]);
            row.push_back(extra["write_pending"]);
            row.push_back(extra["scan_pending"]);
            row.push_back(NumberToString(infos[i].tablet_onload()));
            row.push_back(NumberToString(infos[i].tablet_onbusy()));
            row.push_back(utils::ConvertByteToString(infos[i].mem_used()));
            row.push_back(NumberToString(infos[i].cpu_usage()));
            row.push_back(utils::ConvertByteToString(infos[i].net_tx()));
            row.push_back(utils::ConvertByteToString(infos[i].net_rx()));
            row.push_back(utils::ConvertByteToString(infos[i].dfs_io_r()));
            row.push_back(utils::ConvertByteToString(infos[i].dfs_io_w()));
            printer.AddRow(row);
        }
    } else {
        cols = 7;
        printer.Reset(cols);
        printer.AddRow(cols,
                       " ", "address", "status", "workload",
                       "tablet", "load", "busy");
        std::vector<string> row;
        for (size_t i = 0; i < infos.size(); ++i) {
            row.clear();
            row.push_back(NumberToString(i));
            row.push_back(infos[i].addr());
            row.push_back(infos[i].status_m());
            row.push_back(utils::ConvertByteToString(infos[i].load()));
            row.push_back(NumberToString(infos[i].tablet_total()));
            row.push_back(NumberToString(infos[i].tablet_onload()));
            row.push_back(NumberToString(infos[i].tablet_onbusy()));
            printer.AddRow(row);
        }
{'fraction_non_alphanumeric': 0.14223335635601653, 'fraction_numerical': 0.012572726314926893, 'mean_word_length': 1.6365432098765431, 'pattern_counts': {'":': 10, '<': 857, '<?xml version=': 0, '>': 323, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 226, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1314362', 'n_tokens_mistral': 26607, 'n_tokens_neox': 25697, 'n_words': 5441}