<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/util/io/fileutil.h>
#include <eventql/db/Partition.h>
#include <eventql/db/LSMPartitionWriter.h>
#include <eventql/db/LSMTableIndex.h>
#include <eventql/db/LSMPartitionReader.h>
#include <eventql/db/metadata_operation.h>
#include <eventql/db/metadata_coordinator.h>
#include <eventql/db/server_allocator.h>
#include <eventql/util/protobuf/msg.h>
#include <eventql/util/logging.h>
#include <eventql/util/wallclock.h>
#include <eventql/util/logging.h>
#include <eventql/util/protobuf/MessageDecoder.h>
#include <eventql/io/cstable/RecordShredder.h>
#include <eventql/io/cstable/cstable_writer.h>

#include "eventql/eventql.h"

namespace eventql {

LSMPartitionWriter::LSMPartitionWriter(
    ServerCfg* cfg,
    RefPtr<Partition> partition,
    PartitionSnapshotRef* head) :
    PartitionWriter(head),
    partition_(partition),
    compaction_strategy_(
        new SimpleCompactionStrategy(
            partition_,
            cfg->idx_cache.get())),
    idx_cache_(cfg->idx_cache.get()),
    cdir_(cfg->config_directory),
    repl_(cfg->repl_scheme.get()),
    partition_split_threshold_(kDefaultPartitionSplitThresholdBytes) {}

Set<SHA1Hash> LSMPartitionWriter::insertRecords(
    const ShreddedRecordList& records) {
  std::unique_lock<std::mutex> lk(mutex_);
  if (frozen_) {
    RAISE(kIllegalStateError, "partition is frozen");
  }

  auto snap = head_->getSnapshot();
  const auto& tables = snap->state.lsm_tables();
  if (tables.size() > kMaxLSMTables) {
    RAISE(kRuntimeError, "partition is overloaded, can't insert");
  }

  logTrace(
      "tsdb",
      "Insert $0 record into partition $1/$2/$3",
      records.getNumRecords(),
      snap->state.tsdb_namespace(),
      snap->state.table_key(),
      snap->key.toString());

  HashMap<SHA1Hash, uint64_t> rec_versions;
  for (size_t i = 0; i < records.getNumRecords(); ++i) {
    const auto& record_id = records.getRecordID(i);

    rec_versions.emplace(
        record_id,
        snap->head_arena->fetchRecordVersion(record_id));
  }

  if (snap->compacting_arena.get() != nullptr) {
    for (auto& r : rec_versions) {
      auto v = snap->compacting_arena->fetchRecordVersion(r.first);
      if (v > r.second) {
        r.second = v;
      }
    }
  }

  for (auto tbl = tables.rbegin(); tbl != tables.rend(); ++tbl) {
    auto idx = idx_cache_->lookup(
        FileUtil::joinPaths(snap->rel_path, tbl->filename()));
    idx->lookup(&rec_versions);

    // FIMXE early exit...
  }

  Vector<bool> record_flags_skip(records.getNumRecords(), false);
  Vector<bool> record_flags_update(records.getNumRecords(), false);

  if (!rec_versions.empty()) {
    for (size_t i = 0; i < records.getNumRecords(); ++i) {
      const auto& record_id = records.getRecordID(i);
      auto headv = rec_versions[record_id];
      if (headv > 0) {
        record_flags_update[i] = true;
      }<commit_msg>opportunistically fetch indexes before grabbing write lock<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/util/io/fileutil.h>
#include <eventql/db/Partition.h>
#include <eventql/db/LSMPartitionWriter.h>
#include <eventql/db/LSMTableIndex.h>
#include <eventql/db/LSMPartitionReader.h>
#include <eventql/db/metadata_operation.h>
#include <eventql/db/metadata_coordinator.h>
#include <eventql/db/server_allocator.h>
#include <eventql/util/protobuf/msg.h>
#include <eventql/util/logging.h>
#include <eventql/util/wallclock.h>
#include <eventql/util/logging.h>
#include <eventql/util/protobuf/MessageDecoder.h>
#include <eventql/io/cstable/RecordShredder.h>
#include <eventql/io/cstable/cstable_writer.h>

#include "eventql/eventql.h"

namespace eventql {

LSMPartitionWriter::LSMPartitionWriter(
    ServerCfg* cfg,
    RefPtr<Partition> partition,
    PartitionSnapshotRef* head) :
    PartitionWriter(head),
    partition_(partition),
    compaction_strategy_(
        new SimpleCompactionStrategy(
            partition_,
            cfg->idx_cache.get())),
    idx_cache_(cfg->idx_cache.get()),
    cdir_(cfg->config_directory),
    repl_(cfg->repl_scheme.get()),
    partition_split_threshold_(kDefaultPartitionSplitThresholdBytes) {}

Set<SHA1Hash> LSMPartitionWriter::insertRecords(
    const ShreddedRecordList& records) {
  HashMap<SHA1Hash, uint64_t> rec_versions;
  for (size_t i = 0; i < records.getNumRecords(); ++i) {
    rec_versions.emplace(records.getRecordID(i), 0);
  }

  // opportunistically fetch indexes before going into critical section
  auto snap = head_->getSnapshot();
  Set<String> prepared_indexes;
  {
    const auto& tables = snap->state.lsm_tables();
    for (auto tbl = tables.rbegin(); tbl != tables.rend(); ++tbl) {
      auto idx_path = FileUtil::joinPaths(snap->rel_path, tbl->filename());
      auto idx = idx_cache_->lookup(idx_path);
      idx->lookup(&rec_versions);
      prepared_indexes.insert(idx_path);
    }
  }

  std::unique_lock<std::mutex> lk(mutex_);
  if (frozen_) {
    RAISE(kIllegalStateError, "partition is frozen");
  }

  snap = head_->getSnapshot();
  const auto& tables = snap->state.lsm_tables();
  if (tables.size() > kMaxLSMTables) {
    RAISE(kRuntimeError, "partition is overloaded, can't insert");
  }

  logTrace(
      "tsdb",
      "Insert $0 record into partition $1/$2/$3",
      records.getNumRecords(),
      snap->state.tsdb_namespace(),
      snap->state.table_key(),
      snap->key.toString());

  if (snap->compacting_arena.get() != nullptr) {
    for (auto& r : rec_versions) {
      auto v = snap->compacting_arena->fetchRecordVersion(r.first);
      if (v > r.second) {
        r.second = v;
      }
    }
  }

  for (auto tbl = tables.rbegin(); tbl != tables.rend(); ++tbl) {
    auto idx_path = FileUtil::joinPaths(snap->rel_path, tbl->filename());
    if (prepared_indexes.count(idx_path) > 0) {
      continue;
    }

    auto idx = idx_cache_->lookup(idx_path);
    idx->lookup(&rec_versions);
  }

  Vector<bool> record_flags_skip(records.getNumRecords(), false);
  Vector<bool> record_flags_update(records.getNumRecords(), false);

  if (!rec_versions.empty()) {
    for (size_t i = 0; i < records.getNumRecords(); ++i) {
      const auto& record_id = records.getRecordID(i);
      auto headv = rec_versions[record_id];
      if (headv > 0) {
        record_flags_update[i] = true;
      }
{'fraction_non_alphanumeric': 0.12168746286393345, 'fraction_numerical': 0.0042780748663101605, 'mean_word_length': 3.9158878504672896, 'pattern_counts': {'":': 0, '<': 60, '<?xml version=': 0, '>': 103, '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': '4806824', 'n_tokens_mistral': 2669, 'n_tokens_neox': 2586, 'n_words': 696}