<commit_before>/**
 * Copyright (c) 2016 zScale Technology GmbH <<EMAIL>>
 * Authors:
 *   - <NAME> <<EMAIL>>
 *
 * This program is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Affero General Public License ("the license") as
 * published by the Free Software Foundation, either version 3 of the License,
 * or any later version.
 *
 * In accordance with Section 7(e) of the license, the licensing of the Program
 * under the license does not imply a trademark license. Therefore any rights,
 * title and interest in our trademarks remain entirely with us.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the license for more details.
 *
 * You can be released from the requirements of the license by purchasing a
 * commercial license. Buying such a license is mandatory as soon as you develop
 * commercial activities involving this program without disclosing the source
 * code of your own applications
 */
#include <eventql/db/LSMPartitionReplication.h>
#include <eventql/db/LSMPartitionReader.h>
#include <eventql/db/LSMPartitionWriter.h>
#include <eventql/db/ReplicationScheme.h>
#include <eventql/db/ReplicationWorker.h>
#include <eventql/util/logging.h>
#include <eventql/util/io/fileutil.h>
#include <eventql/util/protobuf/msg.h>
#include <eventql/util/protobuf/MessageEncoder.h>
#include <eventql/util/protobuf/MessagePrinter.h>
#include <eventql/io/cstable/RecordMaterializer.h>
#include <eventql/db/metadata_operations.pb.h>
#include <eventql/db/metadata_coordinator.h>

#include "eventql/eventql.h"

namespace eventql {

const size_t LSMPartitionReplication::kMaxBatchSizeRows = 1024;
const size_t LSMPartitionReplication::kMaxBatchSizeBytes = 1024 * 1024 * 2; // 2 MB

LSMPartitionReplication::LSMPartitionReplication(
    RefPtr<Partition> partition,
    RefPtr<ReplicationScheme> repl_scheme,
    ConfigDirectory* cdir,
    http::HTTPConnectionPool* http) :
    PartitionReplication(partition, repl_scheme, http), cdir_(cdir) {}

bool LSMPartitionReplication::needsReplication() const {
  // check if we have seen the latest metadata transaction, otherwise enqueue
  if (snap_->state.last_metadata_txnid().empty()) {
    return true;
  }

  SHA1Hash last_seen_txid(
      snap_->state.last_metadata_txnid().data(),
      snap_->state.last_metadata_txnid().size());

  auto last_txid = partition_->getTable()->getLastMetadataTransaction();
  if (last_txid.getTransactionID() != last_seen_txid) {
    return true;
  }

  // check if we are splitting
  if (snap_->state.is_splitting()) {
    return true;
  }

  // check if we have joining servers
  if (snap_->state.has_joining_servers()) {
    return true;
  }

  // check if all replicas named in the current metadata transaction have seen
  // the latest sequence, otherwise enqueue
  auto& writer = dynamic_cast<LSMPartitionWriter&>(*partition_->getWriter());
  auto repl_state = writer.fetchReplicationState();
  auto head_offset = snap_->state.lsm_sequence();
  for (const auto& r : snap_->state.replication_targets()) {
    auto replica_offset = replicatedOffsetFor(repl_state, r);
    if (replica_offset < head_offset) {
      return true;
    }
  }

  return false;
}

void LSMPartitionReplication::replicateTo(
    const ReplicationTarget& replica,
    uint64_t replicated_offset,
    ReplicationInfo* replication_info) {
  auto server_cfg = cdir_->getServerConfig(replica.server_id());
  if (server_cfg.server_status() != SERVER_UP) {
    RAISE(kRuntimeError, "server is down");
  }

  SHA1Hash target_partition_id(
      replica.partition_id().data(),
      replica.partition_id().size());

  size_t records_sent = 0;
  size_t bytes_sent = 0;

  const auto& tables = snap_->state.lsm_tables();
  for (const auto& tbl : tables) {
    if (tbl.last_sequence() < replicated_offset) {
      continue;
    }

    // open cstable
    auto cstable_file = FileUtil::joinPaths(
        snap_->base_path,
        tbl.filename() + ".cst");
    auto cstable = cstable::CSTableReader::openFile(cstable_file);
    uint64_t nrecs = cstable->numRecords();

    // in batches
    for (uint64_t nrecs_cur = 0; nrecs_cur < nrecs; ) {
      auto upload_batchsize = std::min(nrecs - nrecs_cur, uint64_t(256)); // FIXME
      nrecs_cur += upload_batchsize;
      Vector<bool> upload_skiplist(upload_batchsize, true);
      size_t upload_nskipped = 0;
      ShreddedRecordListBuilder upload_builder;

      // read metadata
      readBatchMetadata(
          cstable.get(),
          upload_batchsize,
          replicated_offset,
          tbl.has_skiplist(),
          replica.keyrange_begin(),
          replica.keyrange_end(),
          &upload_builder,
          &upload_skiplist,
          &upload_nskipped);

      // FIXME: send metadata list to remote and merge skiplist<commit_msg>set repl batch size to 128<commit_after>/**
 * Copyright (c) 2016 zScale Technology GmbH <<EMAIL>>
 * Authors:
 *   - <NAME> <<EMAIL>>
 *
 * This program is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Affero General Public License ("the license") as
 * published by the Free Software Foundation, either version 3 of the License,
 * or any later version.
 *
 * In accordance with Section 7(e) of the license, the licensing of the Program
 * under the license does not imply a trademark license. Therefore any rights,
 * title and interest in our trademarks remain entirely with us.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the license for more details.
 *
 * You can be released from the requirements of the license by purchasing a
 * commercial license. Buying such a license is mandatory as soon as you develop
 * commercial activities involving this program without disclosing the source
 * code of your own applications
 */
#include <eventql/db/LSMPartitionReplication.h>
#include <eventql/db/LSMPartitionReader.h>
#include <eventql/db/LSMPartitionWriter.h>
#include <eventql/db/ReplicationScheme.h>
#include <eventql/db/ReplicationWorker.h>
#include <eventql/util/logging.h>
#include <eventql/util/io/fileutil.h>
#include <eventql/util/protobuf/msg.h>
#include <eventql/util/protobuf/MessageEncoder.h>
#include <eventql/util/protobuf/MessagePrinter.h>
#include <eventql/io/cstable/RecordMaterializer.h>
#include <eventql/db/metadata_operations.pb.h>
#include <eventql/db/metadata_coordinator.h>

#include "eventql/eventql.h"

namespace eventql {

const size_t LSMPartitionReplication::kMaxBatchSizeRows = 1024;
const size_t LSMPartitionReplication::kMaxBatchSizeBytes = 1024 * 1024 * 2; // 2 MB

LSMPartitionReplication::LSMPartitionReplication(
    RefPtr<Partition> partition,
    RefPtr<ReplicationScheme> repl_scheme,
    ConfigDirectory* cdir,
    http::HTTPConnectionPool* http) :
    PartitionReplication(partition, repl_scheme, http), cdir_(cdir) {}

bool LSMPartitionReplication::needsReplication() const {
  // check if we have seen the latest metadata transaction, otherwise enqueue
  if (snap_->state.last_metadata_txnid().empty()) {
    return true;
  }

  SHA1Hash last_seen_txid(
      snap_->state.last_metadata_txnid().data(),
      snap_->state.last_metadata_txnid().size());

  auto last_txid = partition_->getTable()->getLastMetadataTransaction();
  if (last_txid.getTransactionID() != last_seen_txid) {
    return true;
  }

  // check if we are splitting
  if (snap_->state.is_splitting()) {
    return true;
  }

  // check if we have joining servers
  if (snap_->state.has_joining_servers()) {
    return true;
  }

  // check if all replicas named in the current metadata transaction have seen
  // the latest sequence, otherwise enqueue
  auto& writer = dynamic_cast<LSMPartitionWriter&>(*partition_->getWriter());
  auto repl_state = writer.fetchReplicationState();
  auto head_offset = snap_->state.lsm_sequence();
  for (const auto& r : snap_->state.replication_targets()) {
    auto replica_offset = replicatedOffsetFor(repl_state, r);
    if (replica_offset < head_offset) {
      return true;
    }
  }

  return false;
}

void LSMPartitionReplication::replicateTo(
    const ReplicationTarget& replica,
    uint64_t replicated_offset,
    ReplicationInfo* replication_info) {
  auto server_cfg = cdir_->getServerConfig(replica.server_id());
  if (server_cfg.server_status() != SERVER_UP) {
    RAISE(kRuntimeError, "server is down");
  }

  SHA1Hash target_partition_id(
      replica.partition_id().data(),
      replica.partition_id().size());

  size_t records_sent = 0;
  size_t bytes_sent = 0;

  const auto& tables = snap_->state.lsm_tables();
  for (const auto& tbl : tables) {
    if (tbl.last_sequence() < replicated_offset) {
      continue;
    }

    // open cstable
    auto cstable_file = FileUtil::joinPaths(
        snap_->base_path,
        tbl.filename() + ".cst");
    auto cstable = cstable::CSTableReader::openFile(cstable_file);
    uint64_t nrecs = cstable->numRecords();

    // in batches
    for (uint64_t nrecs_cur = 0; nrecs_cur < nrecs; ) {
      auto upload_batchsize = std::min(nrecs - nrecs_cur, uint64_t(128)); // FIXME
      nrecs_cur += upload_batchsize;
      Vector<bool> upload_skiplist(upload_batchsize, true);
      size_t upload_nskipped = 0;
      ShreddedRecordListBuilder upload_builder;

      // read metadata
      readBatchMetadata(
          cstable.get(),
          upload_batchsize,
          replicated_offset,
          tbl.has_skiplist(),
          replica.keyrange_begin(),
          replica.keyrange_end(),
          &upload_builder,
          &upload_skiplist,
          &upload_nskipped);

      // FIXME: send metadata list to remote and merge skiplist
{'fraction_non_alphanumeric': 0.10124531740407006, 'fraction_numerical': 0.007795889440113395, 'mean_word_length': 3.9095427435387675, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 75, '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': '661579', 'n_tokens_mistral': 3097, 'n_tokens_neox': 2973, 'n_words': 822}