LogNexus_compress.cpp

#include <algorithm>
#include <atomic>
#include <cctype>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <cstdlib>
#include <deque>
#include <filesystem>
#include <fstream>
#include <functional>
#include <future>
#include <iomanip>
#include <iostream>
#include <map>
#include <memory>
#include <mutex>
#include <numeric>
#include <queue>
#include <regex>
#include <sstream>
#include <stdexcept>
#include <string>
#include <string_view>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <vector>

#define PCRE2_CODE_UNIT_WIDTH 8
#include <pcre2.h>


namespace fs = std::filesystem;

struct Pcre2Deleter {
  void operator()(pcre2_code *re) const {
    if (re)
      pcre2_code_free(re);
  }
};
using Pcre2UniquePtr = std::unique_ptr<pcre2_code, Pcre2Deleter>;

// ===== Threshold Parameters (for build_variable_tree) =====
double g_threshold_ratio = 0.02;   // Recommended fixed threshold for untuned use
bool g_skip_stage2 = false;        // RQ3 ablation: skip Variable Subtree Encoding
bool g_single_archive_mode = false; // RQ2: single-archive mode (no chunking)
bool g_s1_lzma_mode = false;       // Stage 1 plus generic LZMA ablation
// ============================================================

// Optional counters for the Stage-2 case study. They are disabled unless the
// experiment launcher explicitly sets LOGNEXUS_CASE_STUDY_COUNTERS=1.
std::mutex g_case_study_counter_mutex;

bool case_study_counter_enabled() {
  const char *env = std::getenv("LOGNEXUS_CASE_STUDY_COUNTERS");
  return env && std::string(env) != "0";
}

void append_case_study_counter(const fs::path &base_output_dir,
                               const std::string &dataset_type,
                               int chunk_id, const std::string &mode,
                               unsigned long long log_count,
                               unsigned long long stage1_vars,
                               unsigned long long stage1_var_bytes,
                               unsigned long long stage2_encoded_vars,
                               unsigned long long stage2_encoded_var_bytes,
                               unsigned long long stage3_residual_vars,
                               unsigned long long stage3_residual_var_bytes) {
  std::lock_guard<std::mutex> lock(g_case_study_counter_mutex);
  fs::path counter_path = base_output_dir / "case_study_counters.csv";
  const bool write_header = !fs::exists(counter_path);
  std::ofstream out(counter_path, std::ios::app);
  if (!out.is_open())
    return;
  if (write_header) {
    out << "Dataset,Chunk,Mode,Log_Count,Stage1_Vars,Stage1_Var_Bytes,"
           "Stage2_Encoded_Vars,Stage2_Encoded_Var_Bytes,"
           "Stage3_Residual_Vars,Stage3_Residual_Var_Bytes\n";
  }
  out << dataset_type << "," << chunk_id << "," << mode << "," << log_count
      << "," << stage1_vars << "," << stage1_var_bytes << ","
      << stage2_encoded_vars << "," << stage2_encoded_var_bytes << ","
      << stage3_residual_vars << "," << stage3_residual_var_bytes << "\n";
}

constexpr uintmax_t SMALL_FILE_PACK_THRESHOLD_BYTES = 32 * 1024;
constexpr char SMALL_FILE_PACK_MAGIC[] = {'L', 'N', 'X', 'P', 'K', '1', '\0'};

struct SmallFilePackConfig {
  bool enabled = true;
  bool adaptive = false;
  uintmax_t threshold_bytes = SMALL_FILE_PACK_THRESHOLD_BYTES;
  size_t adaptive_min_files = 64;
  uintmax_t adaptive_min_saved_bytes = 64 * 1024;
};

std::string trim(std::string_view str) {
  size_t first = str.find_first_not_of(" \t\n\r");
  if (std::string::npos == first)
    return "";
  size_t last = str.find_last_not_of(" \t\n\r");
  return std::string(str.substr(first, (last - first + 1)));
}

std::vector<std::string> split(std::string_view s, char delimiter,
                               int maxsplit = -1) {
  std::vector<std::string> tokens;
  std::string token;
  std::istringstream tokenStream{std::string(s)};
  int count = 0;
  while (std::getline(tokenStream, token, delimiter)) {
    if (maxsplit != -1 && count++ >= maxsplit) {
      std::string rest;
      std::getline(tokenStream, rest);
      tokens.back() += delimiter;
      tokens.back() += token;
      if (!rest.empty()) {
        tokens.back() += delimiter;
        tokens.back() += rest;
      }
      break;
    }
    tokens.push_back(token);
  }
  return tokens;
}

std::vector<std::string> split_stream(std::string_view s) {
  std::vector<std::string> tokens;
  std::stringstream ss{std::string(s)};
  std::string token;
  while (ss >> token)
    tokens.push_back(token);
  return tokens;
}

bool has_numbers(std::string_view s) {
  return std::any_of(s.begin(), s.end(), ::isdigit);
}

long long ip_to_int(std::string_view ip_str) {
  std::string s(ip_str);
  std::stringstream ss(s);
  std::string segment;
  std::stringstream padded_ss;
  while (std::getline(ss, segment, '.')) {
    padded_ss << std::setw(3) << std::setfill('0') << segment;
  }
  try {
    return std::stoll(padded_ss.str());
  } catch (const std::exception &) {
    return 0;
  }
}

long long string_to_padded_int(const std::string &input,
                               const std::vector<int> &padding_widths,
                               char delimiter = 0) {
  std::vector<std::string> parts;
  if (delimiter != 0) {
    parts = split(input, delimiter);
  } else {
    std::string current_num;
    for (char c : input) {
      if (isdigit(c))
        current_num += c;
    }
    if (!current_num.empty())
      parts.push_back(current_num);
  }
  std::stringstream ss;
  for (size_t i = 0; i < parts.size(); ++i) {
    int width = (i < padding_widths.size())
                    ? padding_widths[i]
                    : (padding_widths.empty() ? 0 : padding_widths.back());
    ss << std::setw(width) << std::setfill('0') << parts[i];
  }
  try {
    if (ss.str().empty())
      return 0;
    return std::stoll(ss.str());
  } catch (const std::exception &) {
    return 0;
  }
}

long long digit_runs_to_padded_int(const std::string &input,
                                   const std::vector<int> &padding_widths) {
  std::vector<std::string> parts;
  std::string current;
  for (unsigned char c : input) {
    if (std::isdigit(c)) {
      current.push_back(static_cast<char>(c));
    } else if (!current.empty()) {
      parts.push_back(current);
      current.clear();
    }
  }
  if (!current.empty())
    parts.push_back(current);

  std::stringstream ss;
  for (size_t i = 0; i < parts.size(); ++i) {
    int width = (i < padding_widths.size())
                    ? padding_widths[i]
                    : (padding_widths.empty() ? 0 : padding_widths.back());
    ss << std::setw(width) << std::setfill('0') << parts[i];
  }
  try {
    if (ss.str().empty())
      return 0;
    return std::stoll(ss.str());
  } catch (const std::exception &) {
    return 0;
  }
}

long long numeric_string_to_long(const std::string &s) {
  std::string num_str;
  std::copy_if(s.begin(), s.end(), std::back_inserter(num_str), ::isdigit);
  if (num_str.empty())
    return 0;
  try {
    return std::stoll(num_str);
  } catch (const std::exception &) {
    return 0;
  }
}

unsigned long long encode_linux_dt_shape_flag(const std::string &raw_value) {
  auto space_pos = raw_value.find(' ');
  if (space_pos == std::string::npos)
    return 1ULL;
  return space_pos == 1 ? 0ULL : 1ULL;
}

bool uses_dt_shape(const std::string &dataset_type) {
  return dataset_type == "Linux" || dataset_type == "Apache" ||
         dataset_type == "Mac" || dataset_type == "Thunderbird";
}

unsigned long long encode_colon_field_widths(const std::string &raw_value) {
  std::vector<unsigned long long> widths;
  unsigned long long current_width = 0;
  bool in_digits = false;
  for (unsigned char c : raw_value) {
    if (std::isdigit(c)) {
      ++current_width;
      in_digits = true;
    } else if (c == ':' && in_digits) {
      widths.push_back(current_width);
      current_width = 0;
      in_digits = false;
    } else {
      current_width = 0;
      in_digits = false;
    }
  }
  if (in_digits)
    widths.push_back(current_width);
  while (widths.size() < 4)
    widths.push_back(0);
  return widths[0] * 1000ULL + widths[1] * 100ULL + widths[2] * 10ULL +
         widths[3];
}

// --- Core data structures ---
struct Record {
  long long line_id;
  std::vector<std::string> vars;
};

struct Node {
  std::vector<std::unique_ptr<Node>> children;
  std::unordered_map<std::string, Node *> child_dict;
  int depth = 0;
  int pos = -1;
  std::string digitOrtoken;
  long long cnt = 0;
  std::string pathID;
  std::vector<Record> log_records;

  Node(std::string token = "", int d = 0, int p = -1)
      : depth(d), pos(p), digitOrtoken(std::move(token)) {}

  void add_child(std::unique_ptr<Node> child_node) {
    child_dict[child_node->digitOrtoken] = child_node.get();
    children.push_back(std::move(child_node));
  }

  Node *find_child(std::string_view token) {
    auto it = child_dict.find(std::string(token));
    return (it != child_dict.end()) ? it->second : nullptr;
  }
};

using ExtractedColumns =
    std::unordered_map<std::string, std::vector<long long>>;

constexpr const char *LINUX_DT_SHAPE_PLACEHOLDER = "<dt_shape>";
constexpr const char *TM_SHAPE_PLACEHOLDER = "<tm_shape>";

struct FinalGroupData {
  std::vector<Record> records;
  std::vector<std::string> template_path;
};

struct UnclassifiedVar {
  int pos;
  std::string token;
};

struct UnclassifiedLogInfo {
  long long line_id;
  std::string path_id;
  std::string tree_node_id;
  std::vector<UnclassifiedVar> unmatched_vars;
};

// --- Encoders ---
std::vector<char> elastic_encoder(long long num) {
  std::vector<char> buffer;
  uint64_t cur = (static_cast<uint64_t>(num) << 1) ^ (num >> 63);
  while (true) {
    if (cur < 0x80) {
      buffer.push_back(static_cast<char>(cur));
      break;
    } else {
      buffer.push_back(static_cast<char>((cur & 0x7f) | 0x80));
      cur >>= 7;
    }
  }
  return buffer;
}

std::vector<char> elastic_encoder_unsigned(unsigned long long num) {
  std::vector<char> buffer;
  unsigned long long cur = num;
  while (true) {
    if (cur < 0x80) {
      buffer.push_back(static_cast<char>(cur));
      break;
    } else {
      buffer.push_back(static_cast<char>((cur & 0x7f) | 0x80));
      cur >>= 7;
    }
  }
  return buffer;
}

constexpr char LINE_PREFIX_MAGIC[] = "LNXLP1";
constexpr char TREE_UNSIGNED_MAGIC[] = "LNXU1";

void write_encoded_unsigned(std::ostream &out, unsigned long long value) {
  auto encoded = elastic_encoder_unsigned(value);
  out.write(encoded.data(), encoded.size());
}

size_t common_prefix_length(std::string_view a, std::string_view b) {
  const size_t n = std::min(a.size(), b.size());
  size_t i = 0;
  while (i < n && a[i] == b[i])
    ++i;
  return i;
}

void write_prefix_encoded_templates(
    const fs::path &path,
    const std::vector<std::pair<std::string, long long>> &templates) {
  std::vector<char> io_buffer(1 << 20);
  std::ofstream out;
  out.rdbuf()->pubsetbuf(io_buffer.data(),
                         static_cast<std::streamsize>(io_buffer.size()));
  out.open(path, std::ios::binary);
  if (!out.is_open())
    return;

  out.write(LINE_PREFIX_MAGIC, sizeof(LINE_PREFIX_MAGIC) - 1);
  std::string_view previous;
  for (const auto &entry : templates) {
    std::string_view line(entry.first);
    const size_t prefix = common_prefix_length(previous, line);
    const size_t suffix_len = line.size() - prefix;
    write_encoded_unsigned(out, static_cast<unsigned long long>(prefix));
    write_encoded_unsigned(out, static_cast<unsigned long long>(suffix_len));
    out.write(line.data() + prefix, suffix_len);
    previous = line;
  }
}

class ThreadPool {
public:
  ThreadPool(size_t);
  template <class F, class... Args>
  auto enqueue(F &&f, Args &&...args)
      -> std::future<typename std::invoke_result<F, Args...>::type>;
  ~ThreadPool();

private:
  std::vector<std::thread> workers;
  std::queue<std::function<void()>> tasks;
  std::mutex queue_mutex;
  std::condition_variable condition;
  bool stop;
};
ThreadPool::ThreadPool(size_t threads) : stop(false) {
  for (size_t i = 0; i < threads; ++i)
    workers.emplace_back([this] {
      for (;;) {
        std::function<void()> task;
        {
          std::unique_lock<std::mutex> lock(this->queue_mutex);
          this->condition.wait(
              lock, [this] { return this->stop || !this->tasks.empty(); });
          if (this->stop && this->tasks.empty())
            return;
          task = std::move(this->tasks.front());
          this->tasks.pop();
        }
        task();
      }
    });
}
template <class F, class... Args>
auto ThreadPool::enqueue(F &&f, Args &&...args)
    -> std::future<typename std::invoke_result<F, Args...>::type> {
  using return_type = typename std::invoke_result<F, Args...>::type;
  auto task = std::make_shared<std::packaged_task<return_type()>>(
      std::bind(std::forward<F>(f), std::forward<Args>(args)...));
  std::future<return_type> res = task->get_future();
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    if (stop)
      throw std::runtime_error("enqueue on stopped ThreadPool");
    tasks.emplace([task]() { (*task)(); });
  }
  condition.notify_one();
  return res;
}
ThreadPool::~ThreadPool() {
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    stop = true;
  }
  condition.notify_all();
  for (std::thread &worker : workers)
    worker.join();
}

// =========================================================================
// Processor class (contains specialized logic)
// =========================================================================
struct RegexPattern {
  Pcre2UniquePtr compiled_regex;
  std::string placeholder;
  RegexPattern(Pcre2UniquePtr regex, std::string ph)
      : compiled_regex(std::move(regex)), placeholder(std::move(ph)) {}
  RegexPattern(const RegexPattern &) = delete;
  RegexPattern &operator=(const RegexPattern &) = delete;
  RegexPattern(RegexPattern &&) = default;
  RegexPattern &operator=(RegexPattern &&) = default;
};

class ChunkProcessor {
public:
  ChunkProcessor(const std::string &dataset_type);
  std::pair<std::unique_ptr<Node>, ExtractedColumns>
  process_log_chunk(int chunk_id, const std::vector<std::string> &log_lines,
                    size_t chunk_size,
                    int stage_1_threads = 4);

private:
  void initialize_maps_and_regex();

  struct SubChunkResult {
    std::unique_ptr<Node> partial_tree;
    ExtractedColumns extracted_columns;
  };

  SubChunkResult
  process_sub_chunk(int sub_chunk_id,
                    const std::vector<std::string> &sub_chunk_lines,
                    size_t base_line_id);

  void merge_extracted_columns(ExtractedColumns &target,
                               ExtractedColumns &&source);

  void iterative_merge(Node *target_root, Node *source_root);

  std::string dataset_type_;
  std::unordered_map<std::string, std::vector<RegexPattern>> REGEX_MAP;
  Pcre2UniquePtr MONTH_REGEX;
  Pcre2UniquePtr DAY_REGEX;
  std::unordered_map<std::string, int> MONTH_MAP;
  std::unordered_map<std::string, int> DAY_MAP;
  Pcre2UniquePtr extraction_regex;
};

ChunkProcessor::ChunkProcessor(const std::string &dataset_type)
    : dataset_type_(dataset_type) {
  initialize_maps_and_regex();
}

void ChunkProcessor::initialize_maps_and_regex() {
  MONTH_MAP = {{"Jan", 1}, {"Feb", 2},  {"Mar", 3},  {"Apr", 4},
               {"May", 5}, {"Jun", 6},  {"Jul", 7},  {"Aug", 8},
               {"Sep", 9}, {"Oct", 10}, {"Nov", 11}, {"Dec", 12}};
  DAY_MAP = {{"Sun", 1}, {"Mon", 2}, {"Tue", 3}, {"Wed", 4},
             {"Thu", 5}, {"Fri", 6}, {"Sat", 7}};

  auto compile_regex = [](const char *pattern) -> Pcre2UniquePtr {
    int error_code;
    PCRE2_SIZE error_offset;
    pcre2_code *re = pcre2_compile((PCRE2_SPTR)pattern, PCRE2_ZERO_TERMINATED,
                                   0, &error_code, &error_offset, nullptr);
    if (!re) {
      PCRE2_UCHAR buffer[256];
      pcre2_get_error_message(error_code, buffer, sizeof(buffer));
      throw std::runtime_error("PCRE2 compilation failed at offset " +
                               std::to_string(error_offset) + ": " +
                               (char *)buffer);
    }
    return Pcre2UniquePtr(re);
  };

  MONTH_REGEX =
      compile_regex(R"(\b(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\b)");
  DAY_REGEX = compile_regex(R"(\b(Sun|Mon|Tue|Wed|Thu|Fri|Sat)\b)");

  std::map<std::string, std::vector<std::pair<const char *, const char *>>>
      raw_regex = {
          {"Android",
           {{R"((?<!\S)(?:\d{1,3}\.){3}\d{1,3}(?!\S))", "<I>"},
            {R"((\d{4})-(\d{2})-(\d{2}) (\d{1,2}):(\d{2}):(\d{1,2}):(\d{3}))", "<YmDtm>"},
            {R"((?<![\d-])(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})(?![:\d]))", "<mDt>"}}},
          {"Apache",
           {{R"((\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}))", "<I>"},
            {R"((\d{1,2}) (\d+):(\d+):(\d+))", "<dt>"}}},
          {"BGL",
           {{R"((\d+)-(\d+)-(\d+)-(\d+)\.(\d+)\.(\d+))", "<E>"},
            {R"((?<![A-Za-z0-9:])(\d{1,2}):(\d{2}):(\d{2})(?![A-Za-z0-9:]))", "<t>"},
            {R"((\d{4})\.(\d{2})\.(\d{2}))", "<D>"},
            {R"(\.(\d{6}))", "<m>"},
            {R"(1\d{9})", "<S>"}}},
          {"Hadoop",
           {{R"((\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+),(\d+))", "<Dtm>"}}},
          {"HDFS", {{R"((\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}))", "<I>"}}},
          {"HealthApp", {{R"((\d+):(\d+):(\d+):(\d+))", "<tm>"}}},
          {"HPC", {{R"(1\d{9})", "<S>"}}},
          {"Linux",
           {{R"((?<!\S)(?:\d{1,3}\.){3}\d{1,3}(?!\S))", "<I>"},
            {R"((\d{1,2}) (\d{2}):(\d{2}):(\d{2}))", "<dt>"},
            {R"(\[(\d+)\]:)", "<P>"}}},
          {"Mac",
           {{R"((\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}))", "<I>"},
            {R"((\d{1,2}) (\d{2}):(\d{2}):(\d{2}))", "<dt>"},
            {R"((\d{2}):(\d{2}):(\d{2})\.(\d{3}))", "<tm>"},
            {R"((\d{2}):(\d{2}):(\d{2}))", "<t>"}}},
          {"OpenSSH",
           {{R"((\d{1,2}) (\d+):(\d+):(\d+))", "<dt>"},
            {R"(sshd\[(\d+)\]:)", "<P>"}}},
          {"OpenStack",
           {{R"((\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}))", "<I>"},
            {R"(\.(\d{4})-(\d{2})-(\d{2})_(\d{2}):(\d{2}):(\d{2}))", "<Dt>"},
            {R"((\d{4})-(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})\.(\d{3}))",
             "<Dtm>"}}},
          {"Proxifier", {{R"((\d+)\.(\d+) (\d+):(\d+):(\d+))", "<mDt>"}}},
          {"Spark",
           {{R"((\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}))", "<I>"},
            {R"((\d{2})\/(\d{2})\/(\d{2}) (\d+):(\d+):(\d+))", "<Dt>"},
            {R"((\d+)\.(\d{1}) MB)", "<M>"},
            {R"((\d+)\.(\d{1}) KB)", "<K>"},
            {R"((\d+)\.(\d{1}) GB)", "<G>"},
            {R"((\d+)\.(\d{1}) B)", "<B>"}}},
          {"Thunderbird",
           {{R"((\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}))", "<I>"},
            {R"((\d{1,2}) (\d+):(\d+):(\d+))", "<dt>"},
            {R"((?<!\S)(\d{1,2}):(\d{2}):(\d{2})(?!\S))", "<t>"},
            {R"((\d{4})\.(\d{2})\.(\d{2}))", "<D>"},
            {R"((?<![A-Za-z0-9_./-])\[(\d+)\]:)", "<P>"},
            {R"(1\d{9})", "<S>"}}},
          {"Windows",
           {{R"((\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+),)", "<Dt>"},
            {R"((\d{2}):(\d{2}):(\d{2})\.(\d{3}))", "<tm>"}}},
          {"Zookeeper",
           {{R"((\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+),(\d+))", "<Dtm>"}}}};

  for (const auto &pair : raw_regex) {
    std::vector<RegexPattern> patterns;
    for (const auto &p : pair.second) {
      patterns.emplace_back(compile_regex(p.first), p.second);
    }
    REGEX_MAP[pair.first] = std::move(patterns);
  }
  extraction_regex = compile_regex(
      R"((?<![A-Za-z0-9+\-])(?:[1-9]\d{1,14})(?![A-Za-z0-9+\-]))");
}

// Sub-chunk processing function
ChunkProcessor::SubChunkResult ChunkProcessor::process_sub_chunk(
    int sub_chunk_id, const std::vector<std::string> &sub_chunk_lines,
    size_t base_line_id) {
  auto partial_tree_root = std::make_unique<Node>();
  ExtractedColumns extracted_columns;

  const std::vector<RegexPattern> *regex_patterns_ptr =
      REGEX_MAP.count(dataset_type_) ? &REGEX_MAP.at(dataset_type_) : nullptr;

  pcre2_match_data *match_data = pcre2_match_data_create(32, nullptr);

  for (size_t local_line_id = 0; local_line_id < sub_chunk_lines.size();
       ++local_line_id) {
    std::string modified_line = trim(sub_chunk_lines[local_line_id]);
    if (modified_line.empty())
      continue;

    if (dataset_type_ == "HDFS") {
      auto tokens = split(modified_line, ' ', 3);
      if (tokens.size() == 4) {
        try {
          extracted_columns["<Da>"].push_back(std::stoll(tokens[0]));
          extracted_columns["<Ti>"].push_back(std::stoll(tokens[1]));
          extracted_columns["<P>"].push_back(std::stoll(tokens[2]));
          modified_line = "<Da> <Ti> <P> " + tokens[3];
        } catch (const std::exception &) {
        }
      }
    } else if (dataset_type_ == "HPC") {
      auto tokens = split(modified_line, ' ', 1);
      if (tokens.size() == 2) {
        try {
          extracted_columns["<Lid>"].push_back(std::stoll(tokens[0]));
          modified_line = "<Lid> " + tokens[1];
        } catch (const std::exception &) {
        }
      }
    }

    std::string temp_line;
    PCRE2_SIZE last_match_end = 0;

    temp_line.clear();
    last_match_end = 0;
    while (pcre2_match(MONTH_REGEX.get(), (PCRE2_SPTR)modified_line.c_str(),
                       modified_line.length(), last_match_end, 0, match_data,
                       nullptr) >= 0) {
      PCRE2_SIZE *ovector = pcre2_get_ovector_pointer(match_data);
      temp_line.append(modified_line, last_match_end,
                       ovector[0] - last_match_end);
      std::string month_str(
          modified_line.substr(ovector[2], ovector[3] - ovector[2]));
      if (MONTH_MAP.count(month_str)) {
        extracted_columns["<X>"].push_back(MONTH_MAP.at(month_str));
        temp_line.append("<X>");
      } else {
        temp_line.append(month_str);
      }
      last_match_end = ovector[1];
    }
    temp_line.append(modified_line, last_match_end, std::string::npos);
    if (last_match_end > 0)
      modified_line = temp_line;

    temp_line.clear();
    last_match_end = 0;
    while (pcre2_match(DAY_REGEX.get(), (PCRE2_SPTR)modified_line.c_str(),
                       modified_line.length(), last_match_end, 0, match_data,
                       nullptr) >= 0) {
      PCRE2_SIZE *ovector = pcre2_get_ovector_pointer(match_data);
      temp_line.append(modified_line, last_match_end,
                       ovector[0] - last_match_end);
      std::string day_str(
          modified_line.substr(ovector[2], ovector[3] - ovector[2]));
      if (DAY_MAP.count(day_str)) {
        extracted_columns["<Y>"].push_back(DAY_MAP.at(day_str));
        temp_line.append("<Y>");
      } else {
        temp_line.append(day_str);
      }
      last_match_end = ovector[1];
    }
    temp_line.append(modified_line, last_match_end, std::string::npos);
    if (last_match_end > 0)
      modified_line = temp_line;

    if (regex_patterns_ptr) {
      for (const auto &pattern : *regex_patterns_ptr) {
        temp_line.clear();
        last_match_end = 0;
        while (pcre2_match(pattern.compiled_regex.get(),
                           (PCRE2_SPTR)modified_line.c_str(),
                           modified_line.length(), last_match_end, 0,
                           match_data, nullptr) >= 0) {
          PCRE2_SIZE *ovector = pcre2_get_ovector_pointer(match_data);
          temp_line.append(modified_line, last_match_end,
                           ovector[0] - last_match_end);
          std::string match_str(
              modified_line.substr(ovector[0], ovector[1] - ovector[0]));

          long long value_to_store = 0;
          if (pattern.placeholder == "<I>") {
            value_to_store = ip_to_int(match_str);
          } else if (dataset_type_ == "HealthApp" &&
                     pattern.placeholder == "<tm>") {
            value_to_store = string_to_padded_int(match_str, {2, 2, 2, 3}, ':');
          } else if (dataset_type_ == "Android" &&
                     pattern.placeholder == "<mDt>") {
            value_to_store = string_to_padded_int(match_str, {2, 2, 2, 2, 2});
          } else if (dataset_type_ == "Android" &&
                     pattern.placeholder == "<YmDtm>") {
            value_to_store =
                digit_runs_to_padded_int(match_str, {4, 2, 2, 2, 2, 2, 3});
          } else {
            value_to_store = numeric_string_to_long(match_str);
          }
          extracted_columns[pattern.placeholder].push_back(value_to_store);
          if (uses_dt_shape(dataset_type_) && pattern.placeholder == "<dt>") {
            extracted_columns[LINUX_DT_SHAPE_PLACEHOLDER].push_back(
                static_cast<long long>(encode_linux_dt_shape_flag(match_str)));
          }
          if (dataset_type_ == "HealthApp" && pattern.placeholder == "<tm>") {
            extracted_columns[TM_SHAPE_PLACEHOLDER].push_back(
                static_cast<long long>(encode_colon_field_widths(match_str)));
          }

          temp_line.append(pattern.placeholder);
          last_match_end = ovector[1];
        }
        temp_line.append(modified_line, last_match_end, std::string::npos);
        if (last_match_end > 0)
          modified_line = temp_line;
      }
    }

    long long global_line_id = base_line_id + local_line_id;

    std::vector<std::string> logmessageL;
    if (dataset_type_ == "HealthApp") {
      auto parts = split(modified_line, '|', 3);
      if (parts.size() >= 4) {
        for (size_t i = 0; i < 3; ++i)
          logmessageL.push_back(trim(parts[i]));
        auto tail_parts = split_stream(parts[3]);
        logmessageL.insert(logmessageL.end(), tail_parts.begin(),
                           tail_parts.end());
      } else {
        logmessageL = split_stream(modified_line);
      }
    } else {
      logmessageL = split_stream(modified_line);
    }

    if (logmessageL.empty())
      continue;

    Node *current_node = partial_tree_root.get();
    std::vector<std::string> initial_vars;

    for (const auto &token : logmessageL) {
      bool is_wildcard = has_numbers(token);
      std::string token_to_find = is_wildcard ? "<*>" : token;

      if (is_wildcard) {
        initial_vars.push_back(token);
      }

      Node *child = current_node->find_child(token_to_find);
      if (!child) {
        auto new_child =
            std::make_unique<Node>(token_to_find, current_node->depth + 1);
        child = new_child.get();
        current_node->add_child(std::move(new_child));
      }
      current_node = child;
      current_node->cnt++;
    }

    current_node->log_records.push_back(
        {global_line_id, std::move(initial_vars)});
  }

  pcre2_match_data_free(match_data);
  return {std::move(partial_tree_root), std::move(extracted_columns)};
}

void ChunkProcessor::merge_extracted_columns(ExtractedColumns &target,
                                             ExtractedColumns &&source) {
  for (auto &[placeholder, values] : source) {
    auto &target_vec = target[placeholder];
    target_vec.insert(target_vec.end(), std::make_move_iterator(values.begin()),
                      std::make_move_iterator(values.end()));
  }
}

void ChunkProcessor::iterative_merge(Node *target_root, Node *source_root) {
  if (!source_root)
    return;

  std::vector<std::pair<Node *, Node *>> stack;
  stack.push_back({target_root, source_root});

  while (!stack.empty()) {
    auto [target_node, source_node] = stack.back();
    stack.pop_back();

    for (auto &source_child_ptr : source_node->children) {
      Node *source_child = source_child_ptr.get();
      Node *target_child = target_node->find_child(source_child->digitOrtoken);

      if (!target_child) {
        auto new_child = std::make_unique<Node>(source_child->digitOrtoken,
                                                target_node->depth + 1);
        target_child = new_child.get();
        target_node->add_child(std::move(new_child));
      }

      target_child->cnt += source_child->cnt;
      if (!source_child->log_records.empty()) {
        target_child->log_records.insert(
            target_child->log_records.end(),
            std::make_move_iterator(source_child->log_records.begin()),
            std::make_move_iterator(source_child->log_records.end()));
      }

      if (!source_child->children.empty()) {
        stack.push_back({target_child, source_child});
      }
    }
  }
}

// === Main processing function: supports intra-chunk parallelism ===
std::pair<std::unique_ptr<Node>, ExtractedColumns>
ChunkProcessor::process_log_chunk(int chunk_id,
                                  const std::vector<std::string> &log_lines,
                                  size_t chunk_size, int stage_1_threads) {
  // If data is too small or single-threaded, process sequentially
  if (log_lines.size() < 10000 || stage_1_threads <= 1) {
    auto result = process_sub_chunk(0, log_lines, chunk_id * chunk_size);
    return {std::move(result.partial_tree),
            std::move(result.extracted_columns)};
  }

  // === Intra-chunk parallel processing ===
  // RQ2: Different sub_chunk_size calculation based on mode
  size_t sub_chunk_size;
  if (g_single_archive_mode) {
    // Single-archive mode: each thread processes fixed 100K lines
    // Exception: if total lines < stage_1_threads * 100K, divide evenly
    const size_t threshold = static_cast<size_t>(stage_1_threads) * 100000;
    if (log_lines.size() < threshold) {
      // Small dataset: divide evenly among threads (same as Chunked Mode)
      sub_chunk_size = (log_lines.size() + stage_1_threads - 1) / stage_1_threads;
    } else {
      // Large dataset: fixed 100K lines per thread to avoid memory explosion
      sub_chunk_size = 100000;
    }
  } else {
    // Normal chunked mode: divide chunk evenly among threads
    sub_chunk_size = (log_lines.size() + stage_1_threads - 1) / stage_1_threads;
  }
  
  const int effective_threads = std::min<int>(
      stage_1_threads, 
      static_cast<int>((log_lines.size() + sub_chunk_size - 1) / sub_chunk_size));
  
  // Calculate total number of sub-chunks
  const size_t total_sub_chunks = (log_lines.size() + sub_chunk_size - 1) / sub_chunk_size;

  ThreadPool pool(effective_threads);
  std::deque<std::future<SubChunkResult>> futures;

  auto final_tree_root = std::make_unique<Node>();
  ExtractedColumns final_extracted_columns;

  auto merge_sub_chunk_result = [&](std::future<SubChunkResult> &future) {
    SubChunkResult result = future.get();
    iterative_merge(final_tree_root.get(), result.partial_tree.get());
    merge_extracted_columns(final_extracted_columns,
                            std::move(result.extracted_columns));
  };

  // Keep at most one wave of sub-chunks resident. This mirrors Denum's
  // bounded block scheduling and prevents large single archives from queuing
  // every 100K-line sub-chunk before merging completed work.
  for (size_t i = 0; i < total_sub_chunks; ++i) {
    size_t start = i * sub_chunk_size;
    if (start >= log_lines.size())
      break;

    size_t end = std::min(start + sub_chunk_size, log_lines.size());
    std::vector<std::string> sub_chunk(log_lines.begin() + start,
                                       log_lines.begin() + end);
    size_t base_line_id = chunk_id * chunk_size + start;

    futures.push_back(pool.enqueue(
        [this, i, sub_chunk = std::move(sub_chunk), base_line_id]() {
          return this->process_sub_chunk(static_cast<int>(i), sub_chunk, base_line_id);
        }));

    if (futures.size() >= static_cast<size_t>(effective_threads)) {
      try {
        merge_sub_chunk_result(futures.front());
      } catch (const std::exception &e) {
        std::cerr << "[Chunk " << chunk_id
                  << "] Error processing sub-chunk: " << e.what() << std::endl;
      }
      futures.erase(futures.begin());
    }
  }

  for (auto &future : futures) {
    try {
      merge_sub_chunk_result(future);
    } catch (const std::exception &e) {
      std::cerr << "[Chunk " << chunk_id
                << "] Error processing sub-chunk: " << e.what() << std::endl;
    }
  }

  return {std::move(final_tree_root), std::move(final_extracted_columns)};
}

void ensure_directory_exists(const fs::path &path) {
  if (!fs::exists(path)) {
    try {
      fs::create_directories(path);
    } catch (const fs::filesystem_error &e) {
      throw std::runtime_error("Failed to create directory " + path.string() +
                               ": " + e.what());
    }
  }
}

void write_u16_le(std::ostream &out, uint16_t value) {
  char bytes[2] = {static_cast<char>(value & 0xFF),
                   static_cast<char>((value >> 8) & 0xFF)};
  out.write(bytes, sizeof(bytes));
}

void write_u32_le(std::ostream &out, uint32_t value) {
  char bytes[4] = {static_cast<char>(value & 0xFF),
                   static_cast<char>((value >> 8) & 0xFF),
                   static_cast<char>((value >> 16) & 0xFF),
                   static_cast<char>((value >> 24) & 0xFF)};
  out.write(bytes, sizeof(bytes));
}

std::string lower_ascii(std::string value) {
  std::transform(value.begin(), value.end(), value.begin(),
                 [](unsigned char c) { return static_cast<char>(std::tolower(c)); });
  return value;
}

bool env_flag_enabled(const char *name) {
  const char *raw = std::getenv(name);
  if (!raw || raw[0] == '\0')
    return false;
  std::string value = lower_ascii(raw);
  return !(value == "0" || value == "false" || value == "off" ||
           value == "none" || value == "no");
}

bool verbose_chunk_logs() {
  static const bool enabled = env_flag_enabled("LOGNEXUS_VERBOSE_CHUNKS");
  return enabled;
}

uintmax_t env_uintmax(const char *name, uintmax_t fallback) {
  const char *raw = std::getenv(name);
  if (!raw || raw[0] == '\0')
    return fallback;
  char *end = nullptr;
  unsigned long long parsed = std::strtoull(raw, &end, 10);
  if (end == raw)
    return fallback;
  return static_cast<uintmax_t>(parsed);
}

const SmallFilePackConfig &small_file_pack_config() {
  static SmallFilePackConfig config = [] {
    SmallFilePackConfig cfg;
    if (const char *raw = std::getenv("LOGNEXUS_SMALL_FILE_PACK")) {
      std::string mode = lower_ascii(raw);
      if (mode == "0" || mode == "false" || mode == "off" ||
          mode == "none") {
        cfg.enabled = false;
      } else if (mode == "adaptive") {
        cfg.adaptive = true;
      }
    }
    cfg.threshold_bytes = env_uintmax("LOGNEXUS_SMALL_FILE_PACK_THRESHOLD",
                                      cfg.threshold_bytes);
    cfg.adaptive_min_files = static_cast<size_t>(env_uintmax(
        "LOGNEXUS_SMALL_FILE_PACK_MIN_FILES", cfg.adaptive_min_files));
    cfg.adaptive_min_saved_bytes = env_uintmax(
        "LOGNEXUS_SMALL_FILE_PACK_MIN_SAVED", cfg.adaptive_min_saved_bytes);
    return cfg;
  }();
  return config;
}

uintmax_t round_up_tar_block(uintmax_t value) {
  constexpr uintmax_t block = 512;
  return ((value + block - 1) / block) * block;
}

void remove_empty_directories(const fs::path &root) {
  std::vector<fs::path> dirs;
  for (const auto &entry : fs::recursive_directory_iterator(root)) {
    if (entry.is_directory())
      dirs.push_back(entry.path());
  }
  std::sort(dirs.begin(), dirs.end(), [](const fs::path &a, const fs::path &b) {
    return a.generic_string().size() > b.generic_string().size();
  });
  for (const auto &dir : dirs) {
    std::error_code ec;
    if (fs::is_empty(dir, ec))
      fs::remove(dir, ec);
  }
}

void pack_small_files_for_archive(const fs::path &chunk_dir) {
  const auto &config = small_file_pack_config();
  if (!config.enabled)
    return;

  struct PackedFile {
    fs::path path;
    std::string rel_path;
    uintmax_t size = 0;
  };

  std::vector<PackedFile> files;
  for (const auto &entry : fs::recursive_directory_iterator(chunk_dir)) {
    if (!entry.is_regular_file())
      continue;
    if (entry.path().filename() == "packed_small_files.bin")
      continue;
    uintmax_t size = entry.file_size();
    if (size > config.threshold_bytes)
      continue;
    std::string rel_path = entry.path().lexically_relative(chunk_dir).generic_string();
    if (rel_path.empty() || rel_path.size() > UINT16_MAX || size > UINT32_MAX)
      continue;
    files.push_back({entry.path(), rel_path, size});
  }

  if (files.empty())
    return;

  uintmax_t estimated_original_tar_bytes = 0;
  uintmax_t packed_payload_bytes = sizeof(SMALL_FILE_PACK_MAGIC) + 4;
  for (const auto &file : files) {
    estimated_original_tar_bytes += 512 + round_up_tar_block(file.size);
    packed_payload_bytes += 2 + file.rel_path.size() + 4 + file.size;
  }
  const uintmax_t estimated_packed_tar_bytes =
      512 + round_up_tar_block(packed_payload_bytes);
  const uintmax_t estimated_saved_bytes =
      estimated_original_tar_bytes > estimated_packed_tar_bytes
          ? estimated_original_tar_bytes - estimated_packed_tar_bytes
          : 0;
  if (config.adaptive &&
      (files.size() < config.adaptive_min_files ||
       estimated_saved_bytes < config.adaptive_min_saved_bytes)) {
    return;
  }

  std::sort(files.begin(), files.end(), [](const PackedFile &a, const PackedFile &b) {
    return a.rel_path < b.rel_path;
  });

  fs::path pack_path = chunk_dir / "packed_small_files.bin";
  std::ofstream out(pack_path, std::ios::binary);
  if (!out.is_open())
    throw std::runtime_error("failed to create packed_small_files.bin");

  out.write(SMALL_FILE_PACK_MAGIC, sizeof(SMALL_FILE_PACK_MAGIC));
  write_u32_le(out, static_cast<uint32_t>(files.size()));
  std::vector<char> buffer(1 << 20);
  for (const auto &file : files) {
    write_u16_le(out, static_cast<uint16_t>(file.rel_path.size()));
    out.write(file.rel_path.data(), static_cast<std::streamsize>(file.rel_path.size()));
    write_u32_le(out, static_cast<uint32_t>(file.size));

    std::ifstream in(file.path, std::ios::binary);
    if (!in.is_open())
      throw std::runtime_error("failed to read small file for packing: " +
                               file.path.string());
    while (in) {
      in.read(buffer.data(), static_cast<std::streamsize>(buffer.size()));
      std::streamsize bytes_read = in.gcount();
      if (bytes_read > 0)
        out.write(buffer.data(), bytes_read);
    }
  }
  out.close();

  for (const auto &file : files) {
    fs::remove(file.path);
  }
  remove_empty_directories(chunk_dir);
}

bool compress_directory(const fs::path &dir_to_compress,
                        const fs::path &archive_path) {
  pack_small_files_for_archive(dir_to_compress);

  std::string command;
#ifdef _WIN32
  // Windows: use tar.exe (usually provided with Git for Windows)
  command = "tar -C \"" + dir_to_compress.parent_path().string() +
            "\" -cJf \"" + archive_path.string() + "\" \"" +
            dir_to_compress.filename().string() + "\"";
#else
  // Linux/macOS
  command = "tar -C \"" + dir_to_compress.parent_path().string() +
            "\" --sort=name --mtime=@0 --owner=0 --group=0 --numeric-owner "
            "-cJf \"" + archive_path.string() + "\" \"" +
            dir_to_compress.filename().string() + "\" > /dev/null 2>&1";
#endif
  int result = std::system(command.c_str());
  if (result != 0) {
    std::cerr << "  [Chunk] Error: Failed to compress directory '"
              << dir_to_compress << "'." << std::endl;
    return false;
  }
  return true;
}

void printTree(Node *node, int dep, std::ostream &file) {
  if (!node)
    return;
  for (int i = 0; i < dep; ++i)
    file << "\t";
  if (node->depth == 0)
    file << "Root";

  else
    file << node->digitOrtoken;
  if (!node->pathID.empty())
    file << " $" << node->pathID;
  file << "\n";

  for (const auto &child : node->children) {
    printTree(child.get(), dep + 1, file);
  }
}

// --- Stage 2: 同构合并 (与原版对齐) ---
void iterative_merge(Node *target_root, Node *source_root);
std::string calculate_subtree_signature(Node *node);
void update_descendant_vars(Node *start_node,
                            const std::string &token_to_insert);

void merge_isomorphic_subtrees(Node *node) {
  if (!node)
    return;
  std::vector<Node *> traversal_stack;
  std::unordered_set<Node *> visited;
  traversal_stack.push_back(node);

  while (!traversal_stack.empty()) {
    Node *current_node = traversal_stack.back();
    if (visited.find(current_node) == visited.end()) {
      visited.insert(current_node);
      for (int i = current_node->children.size() - 1; i >= 0; --i) {
        traversal_stack.push_back(current_node->children[i].get());
      }
    } else {
      traversal_stack.pop_back();
      if (current_node->children.empty())
        continue;

      // --- 计算每个子节点的签名 ---
      std::map<std::string, std::vector<Node *>> signatures_to_nodes;
      for (const auto &child_ptr : current_node->children) {
        Node *child = child_ptr.get();
        std::string signature;
        if (child->children.empty()) {
          signature = "LEAF_NODE";
        } else {
          std::vector<std::string> grandchild_signatures;
          for (const auto &grandchild : child->children) {
            grandchild_signatures.push_back(
                calculate_subtree_signature(grandchild.get()));
          }
          std::sort(grandchild_signatures.begin(), grandchild_signatures.end());
          std::stringstream ss;
          for (const auto &sig : grandchild_signatures)
            ss << sig;
          signature = ss.str();
        }
        signatures_to_nodes[signature].push_back(child);
      }

      std::vector<Node *> final_child_raw_ptrs;

      // --- 关键修改：只处理已有通配符的同构组 ---
      for (auto const &[sig, group] : signatures_to_nodes) {
        if (group.size() > 1) {
          // 分离通配符节点和其他节点
          std::vector<Node *> wildcard_nodes, other_nodes;
          for (Node *n : group) {
            if (n->digitOrtoken == "<*>")
              wildcard_nodes.push_back(n);
            else
              other_nodes.push_back(n);
          }

          // !!! 核心变更：只有在存在通配符节点时才合并 !!!
          if (!wildcard_nodes.empty()) {
            // 选择第一个通配符节点作为合并目标
            Node *wildcard_node = wildcard_nodes[0];

            // 合并其他通配符节点到第一个通配符节点
            for (size_t i = 1; i < wildcard_nodes.size(); ++i) {
              Node *other_wildcard = wildcard_nodes[i];
              wildcard_node->cnt += other_wildcard->cnt;
              if (!other_wildcard->log_records.empty()) {
                wildcard_node->log_records.insert(
                    wildcard_node->log_records.end(),
                    std::make_move_iterator(
                        other_wildcard->log_records.begin()),
                    std::make_move_iterator(other_wildcard->log_records.end()));
              }
              iterative_merge(wildcard_node, other_wildcard);
            }

            // 合并其他同构节点到通配符节点
            for (Node *member_node : other_nodes) {
              update_descendant_vars(member_node, member_node->digitOrtoken);
              wildcard_node->cnt += member_node->cnt;
              if (!member_node->log_records.empty()) {
                wildcard_node->log_records.insert(
                    wildcard_node->log_records.end(),
                    std::make_move_iterator(member_node->log_records.begin()),
                    std::make_move_iterator(member_node->log_records.end()));
              }
              iterative_merge(wildcard_node, member_node);
            }

            // 只保留合并后的通配符节点
            final_child_raw_ptrs.push_back(wildcard_node);
          } else {
            // !!! 关键变更：如果同构组中没有通配符节点，保留所有原始节点 !!!
            final_child_raw_ptrs.insert(final_child_raw_ptrs.end(),
                                        group.begin(), group.end());
          }
        } else {
          // 单独的节点直接保留
          final_child_raw_ptrs.insert(final_child_raw_ptrs.end(), group.begin(),
                                      group.end());
        }
      }

      // --- 重建子节点列表 ---
      std::vector<std::unique_ptr<Node>> final_children_list;
      std::unordered_set<Node *> final_ptr_set(final_child_raw_ptrs.begin(),
                                               final_child_raw_ptrs.end());
      for (auto &child_ptr : current_node->children) {
        if (child_ptr && final_ptr_set.count(child_ptr.get())) {
          final_children_list.push_back(std::move(child_ptr));
        }
      }

      // 更新当前节点的子节点和字典
      current_node->children = std::move(final_children_list);
      current_node->child_dict.clear();
      for (const auto &child : current_node->children) {
        current_node->child_dict[child->digitOrtoken] = child.get();
      }
    }
  }
}

// --- Stage 3: 分组 (与原版对齐) ---
void assign_final_path_ids_and_extract_groups(
    Node *node, std::vector<std::string> &current_path,
    std::map<std::string, FinalGroupData> &final_groups,
    std::atomic<long long> &pathIdGenerator) {
  if (node->depth > 0)
    current_path.push_back(node->digitOrtoken);
  if (!node->log_records.empty()) {
    node->pathID = std::to_string(pathIdGenerator++);
    final_groups[node->pathID] = {std::move(node->log_records), current_path};
  }
  for (const auto &child : node->children) {
    assign_final_path_ids_and_extract_groups(child.get(), current_path,
                                             final_groups, pathIdGenerator);
  }
  if (node->depth > 0)
    current_path.pop_back();
}

// 
std::unique_ptr<Node>
build_variable_tree(const std::string &pathID,
                    const std::vector<std::vector<std::string>> &logs,
                    const std::map<int, std::map<std::string, int>> &var_stats,
                    int log_count, std::atomic<long long> &pathIdGenerator);
void printVarTree(Node *node, int dep, std::ostream &file);
std::pair<Node *, std::vector<int>>
traverse_optimized(Node *root, const std::vector<std::string> &varList);
std::string classify_sub_token(std::string_view token_str);
std::pair<std::string, long long> get_classified_token_id(
    const std::string &token, Pcre2UniquePtr &extraction_regex,
    std::unordered_map<std::string, std::vector<long long>> &classified_maps,
    std::unordered_map<std::string, long long> &template_to_id,
    long long &next_template_id);

// =========================================================================
// Core parallel processing function
// =========================================================================
void process_and_compress_chunk(
    int chunk_id, const std::vector<std::string> &log_chunk, size_t block_size,
    const fs::path &base_output_dir, const std::string &dataset_type,
    int stage_1_threads = 4, int stage_2_threads = 4, bool skip_stage2 = false,
    bool s1_lzma_mode = false) {
  auto chunk_start_time = std::chrono::steady_clock::now();

  fs::path chunk_output_dir =
      base_output_dir / ("chunk_" + std::to_string(chunk_id));
  ensure_directory_exists(chunk_output_dir);

  // --- Preprocessing and local tree building (with intra-chunk parallelism) ---
  ChunkProcessor processor(dataset_type);

  // Stage 1: Structural Tree Construction (with intra-chunk parallelism)
  // Parallel Pre-processing & Local Tree Generation & Global Tree Aggregation
  auto [local_root, extracted_columns] = processor.process_log_chunk(
      chunk_id, log_chunk, block_size, stage_1_threads);

  // Isomorphic Subtree Merging
  merge_isomorphic_subtrees(local_root.get());

  // PathID allocation and grouping
  std::map<std::string, FinalGroupData> final_groups;
  std::vector<std::string> current_path_vec;
  std::atomic<long long> pathIdGenerator{0};
  assign_final_path_ids_and_extract_groups(local_root.get(), current_path_vec,
                                           final_groups, pathIdGenerator);

  const bool collect_case_study_counters = case_study_counter_enabled();
  unsigned long long case_log_count = 0;
  unsigned long long case_stage1_vars = 0;
  unsigned long long case_stage1_var_bytes = 0;
  if (collect_case_study_counters) {
    for (const auto &[path_id_str, group_data] : final_groups) {
      (void)path_id_str;
      for (const auto &rec : group_data.records) {
        ++case_log_count;
        case_stage1_vars += rec.vars.size();
        for (const auto &var : rec.vars)
          case_stage1_var_bytes += var.size();
      }
    }
  }

  if (const auto &parent_path = chunk_output_dir.parent_path();
      fs::exists(parent_path) && fs::is_directory(parent_path)) {
    std::ofstream f_init_tree(chunk_output_dir / "initTree_optimized.txt");
    printTree(local_root.get(), 0, f_init_tree);
  }

  if (s1_lzma_mode) {
    std::vector<UnclassifiedLogInfo> s1_records;
    for (const auto &[path_id_str, group_data] : final_groups) {
      for (const auto &rec : group_data.records) {
        std::vector<UnclassifiedVar> raw_vars;
        raw_vars.reserve(rec.vars.size());
        for (size_t i = 0; i < rec.vars.size(); ++i) {
          raw_vars.push_back({static_cast<int>(i), rec.vars[i]});
        }
        s1_records.push_back({
            rec.line_id,
            path_id_str,
            path_id_str,
            std::move(raw_vars)
        });
      }
    }
    std::sort(s1_records.begin(), s1_records.end(),
              [](const auto &a, const auto &b) { return a.line_id < b.line_id; });

    std::ofstream f_tree_node_ids(chunk_output_dir / "tree_node_ids.bin",
                                  std::ios::binary);
    std::ofstream f_raw_vars(chunk_output_dir / "s1_raw_variables.bin",
                             std::ios::binary);
    for (const auto &log_info : s1_records) {
      auto encoded_node_id = elastic_encoder(std::stoll(log_info.tree_node_id));
      f_tree_node_ids.write(encoded_node_id.data(), encoded_node_id.size());

      auto encoded_count =
          elastic_encoder_unsigned(log_info.unmatched_vars.size());
      f_raw_vars.write(encoded_count.data(), encoded_count.size());
      for (const auto &var_info : log_info.unmatched_vars) {
        const auto &token = var_info.token;
        auto encoded_len = elastic_encoder_unsigned(token.size());
        f_raw_vars.write(encoded_len.data(), encoded_len.size());
        f_raw_vars.write(token.data(), static_cast<std::streamsize>(token.size()));
      }
    }
    f_tree_node_ids.close();
    f_raw_vars.close();

    std::map<std::string, long long> delta_state;
    for (auto const &[placeholder, values] : extracted_columns) {
      if (values.empty())
        continue;
      std::string filename = placeholder;
      filename.erase(std::remove(filename.begin(), filename.end(), '<'),
                     filename.end());
      filename.erase(std::remove(filename.begin(), filename.end(), '>'),
                     filename.end());
      std::ofstream file(chunk_output_dir / ("extracted_" + filename + ".bin"),
                         std::ios::binary);
      if (!file.is_open())
        continue;
      long long &last_val = delta_state[placeholder];
      for (long long val : values) {
        bool use_delta = !(placeholder == "<I>" ||
                           placeholder == LINUX_DT_SHAPE_PLACEHOLDER ||
                           placeholder == TM_SHAPE_PLACEHOLDER);
        bool is_unsigned = (placeholder == "<I>" ||
                            placeholder == LINUX_DT_SHAPE_PLACEHOLDER ||
                            placeholder == TM_SHAPE_PLACEHOLDER);
        std::vector<char> encoded =
            use_delta ? elastic_encoder(val - last_val)
                      : (is_unsigned ? elastic_encoder_unsigned(val)
                                     : elastic_encoder(val));
        if (use_delta)
          last_val = val;
        file.write(encoded.data(), encoded.size());
      }
    }

    fs::path compressed_archive_path =
        base_output_dir / ("compressed_" + std::to_string(chunk_id) + ".tar.xz");
    if (!compress_directory(chunk_output_dir, compressed_archive_path)) {
      throw std::runtime_error("failed to compress chunk directory: " +
                               chunk_output_dir.string());
    }
    try {
      fs::remove_all(chunk_output_dir);
    } catch (const fs::filesystem_error &e) {
      std::cerr << "[Chunk " << chunk_id
                << "] Warning: Failed to remove temporary directory "
                << chunk_output_dir << ": " << e.what() << std::endl;
    }

    auto chunk_end_time = std::chrono::steady_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        chunk_end_time - chunk_start_time);
    (void)duration;
    return;
  }

  // Stage 2: Variable Subtree Encoding (with intra-chunk parallelism)
  std::vector<UnclassifiedLogInfo> unclassified_logs;
  std::mutex unclassified_logs_mutex;

  // Create an independent MegaVarTrees.txt for each chunk
  std::ofstream mega_var_tree_file(chunk_output_dir / "MegaVarTrees.txt");
  std::mutex mega_var_tree_mutex;

  // State for variable classification (independent for each chunk)
  std::unordered_map<std::string, std::vector<long long>> classified_maps;
  std::unordered_map<std::string, long long> template_to_id;
  long long next_template_id = 0;
  std::atomic<unsigned long long> case_stage2_encoded_vars{0};
  std::atomic<unsigned long long> case_stage2_encoded_var_bytes{0};
  std::atomic<unsigned long long> case_stage3_residual_vars{0};
  std::atomic<unsigned long long> case_stage3_residual_var_bytes{0};
  Pcre2UniquePtr extraction_regex;
  try {
    int error_code;
    PCRE2_SIZE error_offset;
    extraction_regex.reset(pcre2_compile(
        (PCRE2_SPTR) R"((?<![A-Za-z0-9+\-])(?:[1-9]\d{1,14})(?![A-Za-z0-9+\-]))",
        PCRE2_ZERO_TERMINATED, 0, &error_code, &error_offset, nullptr));
    if (!extraction_regex)
      throw std::runtime_error("Failed to compile extraction regex");
  } catch (...) {
    throw std::runtime_error("Failed to compile extraction regex");
  }

  if (skip_stage2) {
    // RQ3 Ablation: Skip Variable Subtree Encoding
    // Directly pass all variables as unclassified to Stage 3
    for (const auto &[path_id_str, group_data] : final_groups) {
      for (const auto &rec : group_data.records) {
        std::vector<UnclassifiedVar> all_vars;
        for (size_t i = 0; i < rec.vars.size(); ++i) {
          all_vars.push_back({static_cast<int>(i), rec.vars[i]});
          if (collect_case_study_counters) {
            case_stage3_residual_var_bytes.fetch_add(
                rec.vars[i].size(), std::memory_order_relaxed);
          }
        }
        if (collect_case_study_counters) {
          case_stage3_residual_vars.fetch_add(all_vars.size(),
                                              std::memory_order_relaxed);
        }
        unclassified_logs.push_back({
            rec.line_id,
            path_id_str,
            path_id_str,  // tree_node_id = path_id when no var tree
            std::move(all_vars)
        });
      }
    }
    // MegaVarTrees.txt remains empty when Stage 2 is skipped
  } else {
    // Normal Stage 2: Variable Subtree Encoding
    auto process_group = [&](const std::string &path_id_str,
                             const FinalGroupData &group_data) {
      if (group_data.records.empty())
        return;

      std::vector<std::vector<std::string>> logs_original_vars;
      logs_original_vars.reserve(group_data.records.size());
      std::map<int, std::map<std::string, int>> var_stats;
      for (const auto &rec : group_data.records) {
        logs_original_vars.push_back(rec.vars);
        for (size_t i = 0; i < rec.vars.size(); ++i)
          var_stats[i][rec.vars[i]]++;
      }

      auto var_tree_root =
          build_variable_tree(path_id_str, logs_original_vars, var_stats,
                              logs_original_vars.size(), pathIdGenerator);

      if (!var_tree_root->children.empty()) {
        std::stringstream ss;
        ss << path_id_str << "\n";
        printVarTree(var_tree_root.get(), 0, ss);
        ss << "-END-\n";
        std::lock_guard<std::mutex> lock(mega_var_tree_mutex);
        mega_var_tree_file << ss.str();
      }

      std::vector<UnclassifiedLogInfo> local_unclassified_logs;
      local_unclassified_logs.reserve(group_data.records.size());
      unsigned long long local_stage2_encoded_vars = 0;
      unsigned long long local_stage2_encoded_var_bytes = 0;
      unsigned long long local_stage3_residual_vars = 0;
      unsigned long long local_stage3_residual_var_bytes = 0;

      for (size_t i = 0; i < group_data.records.size(); ++i) {
        const auto &rec = group_data.records[i];
        const auto &varList = logs_original_vars[i];
        auto [last_node, matched_indices] =
            traverse_optimized(var_tree_root.get(), varList);

        std::string tree_node_id;
        std::unordered_set<int> matched_set;
        if (!last_node->pathID.empty()) {
          tree_node_id = last_node->pathID;
          matched_set.insert(matched_indices.begin(), matched_indices.end());
        } else {
          tree_node_id = var_tree_root->digitOrtoken;
        }

        std::vector<UnclassifiedVar> unmatched;
        for (size_t idx = 0; idx < varList.size(); ++idx) {
          if (matched_set.find(idx) == matched_set.end()) {
            unmatched.push_back({static_cast<int>(idx), varList[idx]});
            if (collect_case_study_counters) {
              ++local_stage3_residual_vars;
              local_stage3_residual_var_bytes += varList[idx].size();
            }
          } else if (collect_case_study_counters) {
            ++local_stage2_encoded_vars;
            local_stage2_encoded_var_bytes += varList[idx].size();
          }
        }

        local_unclassified_logs.push_back(
            {rec.line_id, path_id_str, tree_node_id, std::move(unmatched)});
      }

      {
        std::lock_guard<std::mutex> lock(unclassified_logs_mutex);
        unclassified_logs.insert(
            unclassified_logs.end(),
            std::make_move_iterator(local_unclassified_logs.begin()),
            std::make_move_iterator(local_unclassified_logs.end()));
      }
      if (collect_case_study_counters) {
        case_stage2_encoded_vars.fetch_add(local_stage2_encoded_vars,
                                           std::memory_order_relaxed);
        case_stage2_encoded_var_bytes.fetch_add(local_stage2_encoded_var_bytes,
                                                std::memory_order_relaxed);
        case_stage3_residual_vars.fetch_add(local_stage3_residual_vars,
                                            std::memory_order_relaxed);
        case_stage3_residual_var_bytes.fetch_add(
            local_stage3_residual_var_bytes, std::memory_order_relaxed);
      }
    };

    if (stage_2_threads <= 1 || final_groups.size() <= 1) {
      for (const auto &[path_id_str, group_data] : final_groups) {
        process_group(path_id_str, group_data);
      }
    } else {
      const int effective_stage_2_threads = std::min<size_t>(
          static_cast<size_t>(stage_2_threads), final_groups.size());
      ThreadPool pool(effective_stage_2_threads);
      std::vector<std::future<void>> futures;
      futures.reserve(final_groups.size());

      for (const auto &[path_id_str, group_data] : final_groups) {
        futures.emplace_back(pool.enqueue(
            [&, path_id_str]() { process_group(path_id_str, group_data); }));
      }

      for (auto &future : futures) {
        future.get();
      }
    }
  }
  mega_var_tree_file.close();
  if (collect_case_study_counters) {
    append_case_study_counter(
        base_output_dir, dataset_type, chunk_id,
        skip_stage2 ? "S1_S3_LZMA" : "FULL", case_log_count,
        case_stage1_vars, case_stage1_var_bytes,
        case_stage2_encoded_vars.load(std::memory_order_relaxed),
        case_stage2_encoded_var_bytes.load(std::memory_order_relaxed),
        case_stage3_residual_vars.load(std::memory_order_relaxed),
        case_stage3_residual_var_bytes.load(std::memory_order_relaxed));
  }
  final_groups.clear();

  // Stage 3: Residual Data Processing
  // Global Sorting: Sort unclassified logs by line_id
  std::sort(unclassified_logs.begin(), unclassified_logs.end(),
            [](const auto &a, const auto &b) { return a.line_id < b.line_id; });

  std::ofstream f_tree_node_ids(chunk_output_dir / "tree_node_ids.bin",
                                std::ios::binary);
  f_tree_node_ids.write(TREE_UNSIGNED_MAGIC, sizeof(TREE_UNSIGNED_MAGIC) - 1);
  std::map<int, std::vector<std::vector<char>>> columnar_data;
  int max_var_count = 0;

  for (const auto &log_info : unclassified_logs) {
    long long node_id = std::stoll(log_info.tree_node_id);
    if (node_id < 0)
      throw std::runtime_error("tree_node_id cannot be negative");
    auto encoded_node_id =
        elastic_encoder_unsigned(static_cast<unsigned long long>(node_id));
    f_tree_node_ids.write(encoded_node_id.data(), encoded_node_id.size());

    // ResVar Templatization: Convert unmatched variables to template IDs
    std::vector<std::vector<char>> classified_vars_encoded;
    for (const auto &var_info : log_info.unmatched_vars) {
      auto [type, template_id] = get_classified_token_id(
          var_info.token, extraction_regex, classified_maps, template_to_id,
          next_template_id);
      auto encoded_value = elastic_encoder_unsigned(template_id);
      classified_vars_encoded.push_back(std::move(encoded_value));
    }

    // Record the number of unmatched variables in the current log and update the global maximum variable count
    int var_count = classified_vars_encoded.size();
    max_var_count = std::max(max_var_count, var_count);
    // Store encoded variables in corresponding columns (idx is the order index of the variable in the list)
    for (int idx = 0; idx < var_count; ++idx) {
      columnar_data[idx].push_back(std::move(classified_vars_encoded[idx]));
    }
  }
  f_tree_node_ids.close();
  std::vector<UnclassifiedLogInfo>().swap(unclassified_logs);

  std::vector<std::pair<std::string, long long>> sorted_templates(
      template_to_id.begin(), template_to_id.end());
  std::sort(sorted_templates.begin(), sorted_templates.end(),
            [](const auto &a, const auto &b) { return a.second < b.second; });
  write_prefix_encoded_templates(chunk_output_dir / "_var_STRING.txt",
                                 sorted_templates);

  if (verbose_chunk_logs()) {
    std::cout << "[Chunk " << chunk_id
              << "] STRING dictionary has been templated, found "
              << template_to_id.size() << " unique templates." << std::endl;
  }

  for (auto const &[tag, data] : classified_maps) {
    if (data.empty())
      continue;
    std::string filename_part = tag.substr(1, tag.length() - 3);
    std::ofstream f(chunk_output_dir / ("_var_" + filename_part + ".bin"),
                    std::ios::binary);
    if (!f.is_open())
      continue;
    bool use_delta =
        !(tag == "<a_>" || tag == "<b_>" || tag == "<c_>" || tag == "<IP_>" ||
          tag == "<HIP_>");
    long long last_val = 0;
    for (long long value : data) {
      std::vector<char> encoded =
          use_delta ? elastic_encoder(value - last_val)
                    : ((tag == "<IP_>" || tag == "<HIP_>")
                           ? elastic_encoder_unsigned(value)
                           : elastic_encoder(value));
      if (use_delta)
        last_val = value;
      f.write(encoded.data(), encoded.size());
    }
  }

  for (int i = 0; i < max_var_count; ++i) {
    if (columnar_data.count(i)) {
      std::ofstream file(chunk_output_dir / ("MegaVariables_col_" +
                                             std::to_string(i) + ".bin"),
                         std::ios::binary);
      if (!file.is_open())
        continue;
      for (const auto &encoded_data : columnar_data.at(i))
        file.write(encoded_data.data(), encoded_data.size());
    }
  }

  std::map<std::string, long long> delta_state;
  for (auto const &[placeholder, values] : extracted_columns) {
    if (values.empty())
      continue;
    std::string filename = placeholder;
    filename.erase(std::remove(filename.begin(), filename.end(), '<'),
                   filename.end());
    filename.erase(std::remove(filename.begin(), filename.end(), '>'),
                   filename.end());
    std::ofstream file(chunk_output_dir / ("extracted_" + filename + ".bin"),
                       std::ios::binary);
    if (!file.is_open())
      continue;
    long long &last_val = delta_state[placeholder];
    for (long long val : values) {
      bool use_delta = !(placeholder == "<I>" ||
                         placeholder == LINUX_DT_SHAPE_PLACEHOLDER ||
                         placeholder == TM_SHAPE_PLACEHOLDER);
      bool is_unsigned = (placeholder == "<I>" ||
                          placeholder == LINUX_DT_SHAPE_PLACEHOLDER ||
                          placeholder == TM_SHAPE_PLACEHOLDER);
      std::vector<char> encoded =
          use_delta ? elastic_encoder(val - last_val)
                    : (is_unsigned ? elastic_encoder_unsigned(val)
                                   : elastic_encoder(val));
      if (use_delta)
        last_val = val;
      file.write(encoded.data(), encoded.size());
    }
  }

  fs::path compressed_archive_path =
      base_output_dir / ("compressed_" + std::to_string(chunk_id) + ".tar.xz");
  if (!compress_directory(chunk_output_dir, compressed_archive_path)) {
    throw std::runtime_error("failed to compress chunk directory: " +
                             chunk_output_dir.string());
  }
  try {
    fs::remove_all(chunk_output_dir);
  } catch (const fs::filesystem_error &e) {
    std::cerr << "[Chunk " << chunk_id
              << "] Warning: Failed to remove temporary directory "
              << chunk_output_dir << ": " << e.what() << std::endl;
  }

  auto chunk_end_time = std::chrono::steady_clock::now();
  auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
      chunk_end_time - chunk_start_time);
  // std::cout << "[Chunk " << chunk_id << "] Finished. Time: " <<
  // duration.count() << "ms." << std::endl;
}

// =========================================================================
// Main Function - Flow Orchestration
// =========================================================================
int main(int argc, char *argv[]) {
  // Check for --single-archive flag first
  bool single_archive = false;
  bool skip_stage2 = false;
  bool s1_lzma_mode = false;
  for (int i = 1; i < argc; ++i) {
    std::string arg(argv[i]);
    if (arg == "--single-archive") {
      single_archive = true;
      g_single_archive_mode = true;
    } else if (arg == "--skip-stage2") {
      skip_stage2 = true;
      g_skip_stage2 = true;
    } else if (arg == "--s1-lzma") {
      s1_lzma_mode = true;
      g_s1_lzma_mode = true;
    }
  }

  // Different argument requirements based on mode
  int min_args = single_archive ? 4 : 5;  // single-archive doesn't need block_size
  
  if (argc < min_args) {
    std::cerr << "Usage (Chunked Mode):" << std::endl;
    std::cerr << "  " << argv[0]
              << " <log_file> <dataset> <block_size> <num_threads> "
                 "[stage_1_threads] [stage_2_threads] [threshold_ratio] [flags]"
              << std::endl;
    std::cerr << "\nUsage (Single-Archive Mode):" << std::endl;
    std::cerr << "  " << argv[0]
              << " <log_file> <dataset> <num_threads> "
                 "[stage_1_threads] [stage_2_threads] [threshold_ratio] --single-archive"
              << std::endl;
    std::cerr << "\nParameters:" << std::endl;
    std::cerr << "  block_size: lines per chunk (Chunked Mode only)" << std::endl;
    std::cerr << "  num_threads: number of parallel chunks (Chunked) or Stage 1 internal threads (Single-Archive)" << std::endl;
    std::cerr << "  stage_1_threads: threads per chunk for Stage 1 (default: 4)" << std::endl;
    std::cerr << "  stage_2_threads: threads for Stage 2 (default: 4)" << std::endl;
    std::cerr << "  threshold_ratio: variable frequency threshold (default: 0.02)" << std::endl;
    std::cerr << "\nFlags:" << std::endl;
    std::cerr << "  --skip-stage2: RQ3 ablation - skip Variable Subtree Encoding" << std::endl;
    std::cerr << "  --s1-lzma: Stage 1 plus generic LZMA ablation" << std::endl;
    std::cerr << "  --single-archive: RQ2 - process entire file as single chunk" << std::endl;
    return 1;
  }

  const fs::path log_filepath(argv[1]);
  const std::string dataset_type(argv[2]);
  
  // Parse arguments based on mode
  size_t BLOCK_SIZE;
  int num_threads, stage_1_threads, stage_2_threads;
  
  if (single_archive) {
    // Single-Archive Mode: <log_file> <dataset> <num_threads> [s1] [s2] [threshold] --single-archive
    BLOCK_SIZE = static_cast<size_t>(INT_MAX);  // Entire file as one chunk
    num_threads = 1;  // Only one chunk in single-archive mode
    stage_1_threads = std::stoi(argv[3]);  // argv[3] is now stage_1_threads
    stage_2_threads = (argc >= 5 && argv[4][0] != '-') ? std::stoi(argv[4]) : 4;
    
    // Parse threshold_ratio if provided
    for (int i = 3; i < argc; ++i) {
      std::string arg(argv[i]);
      if (arg[0] != '-') {
        try {
          double val = std::stod(arg);
          if (val < 1.0) {  // Looks like a threshold ratio
            g_threshold_ratio = val;
          }
        } catch (...) {}
      }
    }
  } else {
    // Chunked Mode: <log_file> <dataset> <block_size> <num_threads> [s1] [s2] [threshold] [flags]
    BLOCK_SIZE = std::stoul(argv[3]);
    num_threads = std::stoi(argv[4]);
    stage_1_threads = (argc >= 6) ? std::stoi(argv[5]) : 4;
    stage_2_threads = (argc >= 7) ? std::stoi(argv[6]) : 4;
    
    if (argc >= 8 && argv[7][0] != '-') {
      g_threshold_ratio = std::stod(argv[7]);
    }
  }

  // Build output directory name based on flags
  std::string output_suffix = "_lognexus";
  if (single_archive) output_suffix += "_single";
  if (s1_lzma_mode) output_suffix += "_s1_lzma";
  if (skip_stage2) output_suffix += "_no_stage2";
  const fs::path output_dir = "output_" + log_filepath.stem().string() + output_suffix;

  if (!fs::exists(log_filepath)) {
    std::cerr << "Error: Log file not found at '" << log_filepath << "'"
              << std::endl;
    return 1;
  }

  if (fs::exists(output_dir)) {
    try {
      fs::remove_all(output_dir);
    } catch (const fs::filesystem_error &e) {
      std::cerr << "Warning: Failed to remove existing output directory "
                << output_dir << ": " << e.what() << std::endl;
    }
  }
  ensure_directory_exists(output_dir);

  std::cout << "--- LogNexus Compressor ---" << std::endl;
  std::cout << "Log File: " << log_filepath
            << ", Dataset Type: " << dataset_type << std::endl;
  if (single_archive) {
    std::cout << "Mode: SINGLE-ARCHIVE (RQ2)" << std::endl;
    std::cout << "Stage 1 Internal Threads: " << stage_1_threads 
              << " (100K lines per internal thread)" << std::endl;
  } else {
    std::cout << "Block Size: " << BLOCK_SIZE 
              << ", Chunk Threads: " << num_threads << std::endl;
    std::cout << "Stage 1 Threads/Chunk: " << stage_1_threads << std::endl;
  }
  std::cout << "Stage 2 Threads: " << stage_2_threads
            << (skip_stage2 ? " (SKIPPED - RQ3 Ablation)" : "") << std::endl;
  if (s1_lzma_mode) {
    std::cout << "Ablation Mode: S1+LZMA (Stage 2 and Stage 3 disabled; raw residual variables use LZMA backend)" << std::endl;
  }
  std::cout << "Threshold Ratio: " << g_threshold_ratio << std::endl;
  std::cout << "Output Dir: " << output_dir << std::endl;
  std::cout << "--------------------------------------------------------"
            << std::endl;

  auto total_start_time = std::chrono::steady_clock::now();
  long long original_log_size = 0;
  try {
    original_log_size = fs::file_size(log_filepath);
  } catch (const fs::filesystem_error &e) {
    std::cerr << "Error: Cannot get size of log file: " << e.what()
              << std::endl;
    return 1;
  }

  std::ifstream log_file(log_filepath);
  if (!log_file.is_open()) {
    std::cerr << "Error: Cannot open log file." << std::endl;
    return 1;
  }

  int chunk_id_counter = 0;
  {
    ThreadPool pool(num_threads);
    std::deque<std::future<void>> futures;

    auto wait_for_oldest_chunk = [&]() {
      try {
        futures.front().get();
      } catch (const std::exception &e) {
        std::cerr << "An exception occurred in a worker thread: " << e.what()
                  << std::endl;
      }
      futures.erase(futures.begin());
    };

    std::vector<std::string> block_buffer;
    const size_t initial_reserve =
        single_archive ? std::min<size_t>(100000, 1 << 20) : BLOCK_SIZE;
    block_buffer.reserve(initial_reserve);
    std::string line;

    while (std::getline(log_file, line)) {
      block_buffer.push_back(line);
      if (block_buffer.size() >= BLOCK_SIZE) {
        futures.push_back(pool.enqueue(process_and_compress_chunk,
                                       chunk_id_counter++, std::move(block_buffer),
                                       BLOCK_SIZE, output_dir, dataset_type,
                                       stage_1_threads, stage_2_threads,
                                       skip_stage2, s1_lzma_mode));
        block_buffer = std::vector<std::string>();
        block_buffer.reserve(initial_reserve);
        if (futures.size() >= static_cast<size_t>(num_threads)) {
          wait_for_oldest_chunk();
        }
      }
    }

    // Process remaining log lines
    if (!block_buffer.empty()) {
      futures.push_back(pool.enqueue(process_and_compress_chunk,
                                     chunk_id_counter++, std::move(block_buffer),
                                     BLOCK_SIZE, output_dir, dataset_type,
                                     stage_1_threads, stage_2_threads,
                                     skip_stage2, s1_lzma_mode));
      if (futures.size() >= static_cast<size_t>(num_threads)) {
        wait_for_oldest_chunk();
      }
    }

    for (auto &fut : futures) {
      try {
        fut.get();
      } catch (const std::exception &e) {
        std::cerr << "An exception occurred in a worker thread: " << e.what()
                  << std::endl;
      }
    }
  }

  auto total_end_time = std::chrono::steady_clock::now();
  std::cout << "\n--- All chunks processed. ---" << std::endl;

  long long total_compressed_bytes = 0;
  for (int i = 0; i < chunk_id_counter; ++i) {
    fs::path compressed_path =
        output_dir / ("compressed_" + std::to_string(i) + ".tar.xz");
    if (fs::exists(compressed_path)) {
      try {
        total_compressed_bytes += fs::file_size(compressed_path);
      } catch (const fs::filesystem_error &e) {
        std::cerr << "Warning: Could not get size of compressed file "
                  << compressed_path << ": " << e.what() << std::endl;
      }
    }
  }

  auto total_duration = std::chrono::duration_cast<std::chrono::milliseconds>(
      total_end_time - total_start_time);
  double total_duration_sec = total_duration.count() / 1000.0;
  double original_size_mb =
      static_cast<double>(original_log_size) / (1024.0 * 1024.0);
  double compression_speed_mbps =
      (total_duration_sec > 0) ? original_size_mb / total_duration_sec : 0;
  double compression_ratio =
      (total_compressed_bytes > 0)
          ? static_cast<double>(original_log_size) / total_compressed_bytes
          : 0;

  std::cout << "\n--- Final Compression Summary ---" << std::endl;
  std::cout << std::fixed << std::setprecision(3);
  std::cout << "Total time taken:      " << total_duration_sec << " s"
            << std::endl;
  std::cout << "Original log size:     " << original_log_size << " Bytes ("
            << original_size_mb << " MB)" << std::endl;
  std::cout << "Total compressed size: " << total_compressed_bytes << " Bytes"
            << std::endl;
  std::cout << "Compression Speed:     " << compression_speed_mbps << " MB/s"
            << std::endl;
  std::cout << "Compression Ratio:     " << compression_ratio << " : 1"
            << std::endl;
  std::cout << "Total Chunks:          " << chunk_id_counter << std::endl;
  std::cout << "---------------------------------" << std::endl;

  return 0;
}

void iterative_merge(Node *target_root, Node *source_root) {
  if (!source_root)
    return;
  std::vector<std::pair<Node *, Node *>> stack;
  stack.push_back({target_root, source_root});
  while (!stack.empty()) {
    auto [target_node, source_node] = stack.back();
    stack.pop_back();
    for (auto &source_child_ptr : source_node->children) {
      Node *source_child = source_child_ptr.get();
      Node *target_child = target_node->find_child(source_child->digitOrtoken);
      if (!target_child) {
        auto new_child = std::make_unique<Node>(source_child->digitOrtoken,
                                                target_node->depth + 1);
        target_child = new_child.get();
        target_node->add_child(std::move(new_child));
      }
      target_child->cnt += source_child->cnt;
      if (!source_child->log_records.empty()) {
        target_child->log_records.insert(
            target_child->log_records.end(),
            std::make_move_iterator(source_child->log_records.begin()),
            std::make_move_iterator(source_child->log_records.end()));
      }
      if (!source_child->children.empty())
        stack.push_back({target_child, source_child});
    }
  }
}

std::string calculate_subtree_signature(Node *node) {
  if (!node->children.empty()) {
    std::vector<std::string> child_signatures;
    for (const auto &child : node->children)
      child_signatures.push_back(calculate_subtree_signature(child.get()));
    std::sort(child_signatures.begin(), child_signatures.end());
    std::stringstream ss;
    ss << "(" << node->digitOrtoken;
    for (const auto &sig : child_signatures)
      ss << sig;
    ss << ")";
    return ss.str();
  }
  return "(" + node->digitOrtoken + ")";
}

void update_descendant_vars(Node *start_node,
                            const std::string &token_to_insert) {
  std::vector<std::pair<Node *, int>> stack;
  stack.push_back({start_node, 0});
  std::unordered_set<Node *> visited;
  while (!stack.empty()) {
    auto [current_node, wildcard_count] = stack.back();
    stack.pop_back();
    if (visited.count(current_node))
      continue;
    visited.insert(current_node);
    if (!current_node->log_records.empty()) {
      for (auto &record : current_node->log_records) {
        int insertion_index = record.vars.size() - wildcard_count;
        insertion_index = std::max(0, insertion_index);
        record.vars.insert(record.vars.begin() + insertion_index,
                           token_to_insert);
      }
    }
    for (int i = current_node->children.size() - 1; i >= 0; --i) {
      auto &child = current_node->children[i];
      int new_wildcard_count =
          wildcard_count + (child->digitOrtoken == "<*>" ? 1 : 0);
      stack.push_back({child.get(), new_wildcard_count});
    }
  }
}

std::unique_ptr<Node>
build_variable_tree(const std::string &pathID,
                    const std::vector<std::vector<std::string>> &logs,
                    const std::map<int, std::map<std::string, int>> &var_stats,
                    int log_count, std::atomic<long long> &pathIdGenerator) {
  auto tree_root = std::make_unique<Node>(pathID, 0);
  if (log_count == 0)
    return tree_root;
  // Calculate lower_threshold using global threshold_ratio
  int lower_threshold =
      std::max(5, static_cast<int>(log_count * g_threshold_ratio));
  std::map<int, std::map<std::string, int>> filtered_stats;
  std::map<int, int> filtered_unique_var_count, filtered_total_var_count;
  for (const auto &[pos, stats] : var_stats) {
    std::map<std::string, int> pos_filtered;
    int total_count = 0;
    for (const auto &[var, cnt] : stats) {
      if (cnt >= lower_threshold) {
        pos_filtered[var] = cnt;
        total_count += cnt;
      }
    }
    if (!pos_filtered.empty()) {
      filtered_stats[pos] = pos_filtered;
      filtered_unique_var_count[pos] = pos_filtered.size();
      filtered_total_var_count[pos] = total_count;
    }
  }
  if (filtered_stats.empty())
    return tree_root;
  std::vector<int> tree_order;
  for (const auto &[pos, count] : filtered_unique_var_count)
    tree_order.push_back(pos);
  std::sort(tree_order.begin(), tree_order.end(), [&](int posA, int posB) {
    if (filtered_total_var_count.at(posA) != filtered_total_var_count.at(posB))
      return filtered_total_var_count.at(posA) >
             filtered_total_var_count.at(posB);
    return filtered_unique_var_count.at(posA) <
           filtered_unique_var_count.at(posB);
  });
  for (const auto &varList : logs) {
    Node *current_node = tree_root.get();
    for (int level = 0; level < tree_order.size(); ++level) {
      int pos = tree_order[level];
      if (pos < varList.size()) {
        const auto &var_token = varList[pos];
        if (filtered_stats.count(pos) &&
            filtered_stats.at(pos).count(var_token)) {
          Node *child = current_node->find_child(var_token);
          if (!child) {
            auto new_child = std::make_unique<Node>(var_token, level + 1, pos);
            child = new_child.get();
            current_node->add_child(std::move(new_child));
          }
          current_node = child;
          current_node->cnt++;
        } else
          break;
      }
    }
  }
  // Calculate prune_threshold based on threshold_ratio
  double prune_threshold = g_threshold_ratio * log_count;
  std::queue<Node *> q;
  q.push(tree_root.get());
  while (!q.empty()) {
    Node *node = q.front();
    q.pop();
    auto &children = node->children;
    children.erase(std::remove_if(children.begin(), children.end(),
                                  [=](const auto &child) {
                                    return child->cnt < prune_threshold;
                                  }),
                   children.end());
    node->child_dict.clear();
    for (const auto &child : children) {
      node->child_dict[child->digitOrtoken] = child.get();
      q.push(child.get());
    }
  }
  std::function<void(Node *)> assign_pid = [&](Node *node) {
    if (node->depth > 0) {
      long children_cnt_sum = 0;
      for (const auto &child : node->children)
        children_cnt_sum += child->cnt;
      bool is_perfect_passthrough =
          (node->children.size() == 1 && node->children[0]->cnt == node->cnt);
      if (!is_perfect_passthrough &&
          (node->cnt - children_cnt_sum) >= prune_threshold) {
        node->pathID = std::to_string(pathIdGenerator++);
      }
    }
    for (const auto &child : node->children)
      assign_pid(child.get());
  };
  assign_pid(tree_root.get());
  return tree_root;
}

void printVarTree(Node *node, int dep, std::ostream &file) {
  if (!node)
    return;
  if (node->depth == 0) {
    for (const auto &child : node->children)
      printVarTree(child.get(), dep + 1, file);
    return;
  }
  for (int i = 0; i < (dep - 1); ++i)
    file << "\t";
  file << node->digitOrtoken << " $" << node->pos;
  if (!node->pathID.empty())
    file << " " << node->pathID;
  file << "\n";
  for (const auto &child : node->children)
    printVarTree(child.get(), dep + 1, file);
}

std::pair<Node *, std::vector<int>>
traverse_optimized(Node *root, const std::vector<std::string> &varList) {
  Node *cur = root;
  std::vector<int> matched_idxs;
  while (true) {
    bool found_match_in_level = false;
    for (const auto &child_ptr : cur->children) {
      Node *child = child_ptr.get();
      int pos = child->pos;
      if (pos < varList.size() && varList[pos] == child->digitOrtoken) {
        cur = child;
        matched_idxs.push_back(pos);
        found_match_in_level = true;
        break;
      }
    }
    if (!found_match_in_level)
      break;
  }
  return {cur, matched_idxs};
}

std::string classify_sub_token(std::string_view token_str) {
  static const std::regex integer_re(R"(^-?\d+$)");
  static const std::regex ip_addr_re(R"(^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$)");
  static const std::string alpha = "abcdefghijklmnopqrstuvwxyz";
  std::string s_token(token_str);
  if (std::regex_match(s_token, integer_re)) {
    if ((s_token.rfind('0', 0) == 0 && s_token.length() > 1) ||
        (s_token.rfind("-0", 0) == 0 && s_token.length() > 2))
      return "STRING";
    if (s_token.length() >= 15)
      return "STRING";
    try {
      std::string num_part = s_token[0] == '-' ? s_token.substr(1) : s_token;
      if (num_part.empty())
        return "STRING";
      size_t length = num_part.length();
      char len_char = alpha.at(length - 1);
      std::string first_digit_char;
      if (length >= 4)
        first_digit_char = alpha.at(num_part[0] - '0');
      return "<" + std::string(1, len_char) + first_digit_char + "_>";
    } catch (const std::out_of_range &) {
      return "STRING";
    }
  }
  if (std::regex_match(s_token, ip_addr_re))
    return "<IP_>";
  return "STRING";
}

bool try_build_linux_host_ip_template(
    const std::string &token,
    std::unordered_map<std::string, std::vector<long long>> &classified_maps,
    std::unordered_map<std::string, long long> &template_to_id,
    long long &next_template_id, std::string &final_template_str) {
  static const std::regex hip_re(
      R"(^(.*?)(\d{3}\.\d{3}\.\d{3}\.\d{3})(.*)$)");
  std::smatch match;
  if (!std::regex_match(token, match, hip_re))
    return false;

  std::string prefix = match[1].str();
  std::string ip_part = match[2].str();
  std::string suffix = match[3].str();
  if (prefix.empty() && suffix.empty())
    return false;

  auto has_host_context = [](const std::string &s) {
    return std::any_of(s.begin(), s.end(), [](unsigned char c) {
      return std::isalpha(c) || c == '-';
    });
  };
  if (!has_host_context(prefix) && !has_host_context(suffix))
    return false;

  auto template_ref = [&](const std::string &part) -> std::string {
    if (part.empty())
      return "";
    auto it = template_to_id.find(part);
    if (it == template_to_id.end())
      it = template_to_id.emplace(part, next_template_id++).first;
    return "<T:" + std::to_string(it->second) + ">";
  };

  classified_maps["<HIP_>"].push_back(ip_to_int(ip_part));
  final_template_str = template_ref(prefix) + "<HIP_>" + template_ref(suffix);
  return true;
}

std::pair<std::string, long long> get_classified_token_id(
    const std::string &token, Pcre2UniquePtr &extraction_regex,
    std::unordered_map<std::string, std::vector<long long>> &classified_maps,
    std::unordered_map<std::string, long long> &template_to_id,
    long long &next_template_id) {
  std::string final_template_str;
  if (try_build_linux_host_ip_template(token, classified_maps, template_to_id,
                                       next_template_id, final_template_str)) {
    if (template_to_id.find(final_template_str) == template_to_id.end())
      template_to_id[final_template_str] = next_template_id++;
    return {"STRING", template_to_id.at(final_template_str)};
  }
  std::string whole_placeholder = classify_sub_token(token);
  if (whole_placeholder == "<IP_>") {
    classified_maps[whole_placeholder].push_back(ip_to_int(token));
    if (template_to_id.find(whole_placeholder) == template_to_id.end())
      template_to_id[whole_placeholder] = next_template_id++;
    return {"STRING", template_to_id.at(whole_placeholder)};
  }
  std::vector<std::string> parts;
  size_t last_end = 0;
  pcre2_match_data *match_data =
      pcre2_match_data_create_from_pattern(extraction_regex.get(), nullptr);
  if (!match_data)
    throw std::runtime_error("Failed to create PCRE2 match data");
  PCRE2_SIZE start_offset = 0;
  while (start_offset < token.length()) {
    int rc = pcre2_match(extraction_regex.get(), (PCRE2_SPTR)token.c_str(),
                         token.length(), start_offset, 0, match_data, nullptr);
    if (rc < 0)
      break;
    PCRE2_SIZE *ovector = pcre2_get_ovector_pointer(match_data);
    size_t match_start = ovector[0], match_end = ovector[1];
    if (match_start > last_end) {
      std::string prefix = token.substr(last_end, match_start - last_end);
      if (!prefix.empty())
        parts.push_back(prefix);
    }
    std::string extracted_token =
        token.substr(match_start, match_end - match_start);
    std::string placeholder = classify_sub_token(extracted_token);
    if (placeholder != "STRING") {
      long long value_to_store = (placeholder == "<IP_>")
                                     ? ip_to_int(extracted_token)
                                     : std::stoll(extracted_token);
      classified_maps[placeholder].push_back(value_to_store);
      parts.push_back(placeholder);
    } else {
      std::string sub_template = extracted_token;
      if (template_to_id.find(sub_template) == template_to_id.end())
        template_to_id[sub_template] = next_template_id++;
      parts.push_back("<T:" + std::to_string(template_to_id.at(sub_template)) +
                      ">");
    }
    start_offset = match_end;
    last_end = match_end;
  }
  pcre2_match_data_free(match_data);
  if (last_end < token.length()) {
    std::string suffix = token.substr(last_end);
    if (!suffix.empty())
      parts.push_back(suffix);
  }
  std::stringstream ss;
  for (const auto &p : parts)
    if (!p.empty())
      ss << p;
  final_template_str = ss.str();
  if (final_template_str.empty())
    final_template_str = token;
  if (template_to_id.find(final_template_str) == template_to_id.end())
    template_to_id[final_template_str] = next_template_id++;
  return {"STRING", template_to_id.at(final_template_str)};
}