<commit_before>#include "zp_data_table.h"

#include <sys/statvfs.h>
#include <glog/logging.h>

#include "zp_data_server.h"

extern ZPDataServer* zp_data_server;

Table* NewTable(const std::string &table_name, const std::string log_path, const std::string data_path) {
  Table* table = new Table(table_name, log_path, data_path);
  // TODO maybe need check
  return table;
}

//
// Table
//
Table::Table(const std::string& table_name, const std::string &log_path, const std::string &data_path)
  : table_name_(table_name),
  log_path_(log_path),
  data_path_(data_path),
  partition_cnt_(0) {
  if (log_path_.back() != '/') {
    log_path_.push_back('/');
  }
  if (data_path_.back() != '/') {
    data_path_.push_back('/');
  }
  log_path_ += table_name_ + "/";
  data_path_ += table_name_ + "/";

  slash::CreatePath(log_path_);
  slash::CreatePath(data_path_);

  pthread_rwlockattr_t attr;
  pthread_rwlockattr_init(&attr);
  pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
  pthread_rwlock_init(&partition_rw_, &attr);
}

Table::~Table() {
  {
    slash::RWLock l(&partition_rw_, true);
    for (auto iter = partitions_.begin(); iter != partitions_.end(); iter++) {
      delete iter->second;
    }
  }
  pthread_rwlock_destroy(&partition_rw_);
  LOG(INFO) << " Table " << table_name_ << " exit!!!";
}

bool Table::SetPartitionCount(const int count) {
  partition_cnt_ = count;
  DLOG(INFO) << " Set Table: " << table_name_ << " with " << partition_cnt_ << " partitions.";
  return true;
}

Partition* Table::GetPartition(const std::string &key) {
  slash::RWLock l(&partition_rw_, false);
  if (partition_cnt_ > 0) {
    int partition_id = std::hash<std::string>()(key) % partition_cnt_;
    return partitions_[partition_id]; 
  }
  return NULL;
}

Partition* Table::GetPartitionById(const int partition_id) {
  slash::RWLock l(&partition_rw_, false);
  auto it = partitions_.find(partition_id);
  //std::map<int, Partition*>::const_iterator it = partitions_.find(partition_id);
  if (it != partitions_.end()) {
    return it->second;
  }
  return NULL;
}

bool Table::UpdateOrAddPartition(const int partition_id,
    ZPMeta::PState state, const Node& master, const std::set<Node>& slaves) {
  slash::RWLock l(&partition_rw_, true);
  auto iter = partitions_.find(partition_id);
  if (iter != partitions_.end()) {
    //Exist partition: update it
    (iter->second)->Update(state, master, slaves);
    return true;
  }

  if (state != ZPMeta::PState::ACTIVE) {
    DLOG(WARNING) << "New Partition with no active state";
    return false;
  }

  // New Partition
  Partition* partition = NewPartition(table_name_,
      log_path_, data_path_, partition_id, master, slaves);
  assert(partition != NULL);

  partition->Update(ZPMeta::PState::ACTIVE, master, slaves);
  partitions_[partition_id] = partition;

  return true;
}

void Table::LeaveAllPartition() {
  slash::RWLock l(&partition_rw_, false);
  for (auto iter : partitions_) {
    (iter.second)->Leave();
  }
}

uint32_t Table::KeyToPartition(const std::string &key) {
  assert(partition_cnt_ != 0);
  return std::hash<std::string>()(key) % partition_cnt_;
}

void Table::Dump() {
  slash::RWLock l(&partition_rw_, false);
  DLOG(INFO) << "=========================";
  DLOG(INFO) << "    Table : " << table_name_;
  for (auto iter = partitions_.begin(); iter != partitions_.end(); iter++) {
    iter->second->Dump();
  }
  DLOG(INFO) << "--------------------------";
}


void Table::DoTimingTask() {
  slash::RWLock l(&partition_rw_, false);
  for (auto pair : partitions_) {
    pair.second->DoTimingTask();
  }
}

void Table::DumpPartitionBinlogOffsets(std::vector<PartitionBinlogOffset> &offset) {
  slash::RWLock l(&partition_rw_, false);
  PartitionBinlogOffset tboffset;
  for (auto& pair : partitions_) {
    tboffset.partition_id = pair.first;
    (pair.second)->GetBinlogOffset(&(tboffset.filenum), &(tboffset.offset));
    offset.push_back(tboffset);
  }
}

uint64_t Df(const std::string& path) {
  struct statvfs sfs;
  if (statvfs(path.data(), &sfs) != -1) {
    return (uint64_t)(sfs.f_bsize * sfs.f_blocks);
  }
  return 0LL;
}

void Table::GetCapacity(Statistic *stat) {
  stat->Reset();
  stat->table_name = table_name_;
  stat->used_disk = slash::Du(data_path_) + slash::Du(log_path_);
  // TODO anan: we will support table based disk; 
  // For now, just use node's free disk instead.
  stat->free_disk = Df(data_path_);
  DLOG(INFO) << "GetCapacity for table " << table_name_ << ":";
  stat->Dump();
}
<commit_msg>fix invaild partition problem<commit_after>#include "zp_data_table.h"

#include <sys/statvfs.h>
#include <glog/logging.h>

#include "zp_data_server.h"

extern ZPDataServer* zp_data_server;

Table* NewTable(const std::string &table_name, const std::string log_path, const std::string data_path) {
  Table* table = new Table(table_name, log_path, data_path);
  // TODO maybe need check
  return table;
}

//
// Table
//
Table::Table(const std::string& table_name, const std::string &log_path, const std::string &data_path)
  : table_name_(table_name),
  log_path_(log_path),
  data_path_(data_path),
  partition_cnt_(0) {
  if (log_path_.back() != '/') {
    log_path_.push_back('/');
  }
  if (data_path_.back() != '/') {
    data_path_.push_back('/');
  }
  log_path_ += table_name_ + "/";
  data_path_ += table_name_ + "/";

  slash::CreatePath(log_path_);
  slash::CreatePath(data_path_);

  pthread_rwlockattr_t attr;
  pthread_rwlockattr_init(&attr);
  pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
  pthread_rwlock_init(&partition_rw_, &attr);
}

Table::~Table() {
  {
    slash::RWLock l(&partition_rw_, true);
    for (auto iter = partitions_.begin(); iter != partitions_.end(); iter++) {
      delete iter->second;
    }
  }
  pthread_rwlock_destroy(&partition_rw_);
  LOG(INFO) << " Table " << table_name_ << " exit!!!";
}

bool Table::SetPartitionCount(const int count) {
  partition_cnt_ = count;
  DLOG(INFO) << " Set Table: " << table_name_ << " with " << partition_cnt_ << " partitions.";
  return true;
}

Partition* Table::GetPartition(const std::string &key) {
  slash::RWLock l(&partition_rw_, false);
  if (partition_cnt_ > 0) {
    int partition_id = std::hash<std::string>()(key) % partition_cnt_;
    auto it = partitions_.find(partition_id);
    if (it != partitions_.end()) {
      return it->second;
    }
  }
  return NULL;
}

Partition* Table::GetPartitionById(const int partition_id) {
  slash::RWLock l(&partition_rw_, false);
  auto it = partitions_.find(partition_id);
  if (it != partitions_.end()) {
    return it->second;
  }
  return NULL;
}

bool Table::UpdateOrAddPartition(const int partition_id,
    ZPMeta::PState state, const Node& master, const std::set<Node>& slaves) {
  slash::RWLock l(&partition_rw_, true);
  auto iter = partitions_.find(partition_id);
  if (iter != partitions_.end()) {
    //Exist partition: update it
    (iter->second)->Update(state, master, slaves);
    return true;
  }

  if (state != ZPMeta::PState::ACTIVE) {
    DLOG(WARNING) << "New Partition with no active state";
    return false;
  }

  // New Partition
  Partition* partition = NewPartition(table_name_,
      log_path_, data_path_, partition_id, master, slaves);
  assert(partition != NULL);

  partition->Update(ZPMeta::PState::ACTIVE, master, slaves);
  partitions_[partition_id] = partition;

  return true;
}

void Table::LeaveAllPartition() {
  slash::RWLock l(&partition_rw_, false);
  for (auto iter : partitions_) {
    (iter.second)->Leave();
  }
}

uint32_t Table::KeyToPartition(const std::string &key) {
  assert(partition_cnt_ != 0);
  return std::hash<std::string>()(key) % partition_cnt_;
}

void Table::Dump() {
  slash::RWLock l(&partition_rw_, false);
  DLOG(INFO) << "=========================";
  DLOG(INFO) << "    Table : " << table_name_;
  for (auto iter = partitions_.begin(); iter != partitions_.end(); iter++) {
    iter->second->Dump();
  }
  DLOG(INFO) << "--------------------------";
}


void Table::DoTimingTask() {
  slash::RWLock l(&partition_rw_, false);
  for (auto pair : partitions_) {
    pair.second->DoTimingTask();
  }
}

void Table::DumpPartitionBinlogOffsets(std::vector<PartitionBinlogOffset> &offset) {
  slash::RWLock l(&partition_rw_, false);
  PartitionBinlogOffset tboffset;
  for (auto& pair : partitions_) {
    tboffset.partition_id = pair.first;
    (pair.second)->GetBinlogOffset(&(tboffset.filenum), &(tboffset.offset));
    offset.push_back(tboffset);
  }
}

uint64_t Df(const std::string& path) {
  struct statvfs sfs;
  if (statvfs(path.data(), &sfs) != -1) {
    return (uint64_t)(sfs.f_bsize * sfs.f_blocks);
  }
  return 0LL;
}

void Table::GetCapacity(Statistic *stat) {
  stat->Reset();
  stat->table_name = table_name_;
  stat->used_disk = slash::Du(data_path_) + slash::Du(log_path_);
  // TODO anan: we will support table based disk; 
  // For now, just use node's free disk instead.
  stat->free_disk = Df(data_path_);
  DLOG(INFO) << "GetCapacity for table " << table_name_ << ":";
  stat->Dump();
}

{'fraction_non_alphanumeric': 0.16923925027563397, 'fraction_numerical': 0.0024255788313120178, 'mean_word_length': 4.4382494004796165, 'pattern_counts': {'":': 2, '<': 80, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1043375', 'n_tokens_mistral': 3206, 'n_tokens_neox': 3146, 'n_words': 629}