24057299873

Committed 07 Apr 2026 12:01AM UTC coverage: 52.076% (+0.8%) from 51.228%

Build # 24057299873

Build Type

push

github

Committed by

rayandrew

Commit Message

feat(rocksdb): migrate SQLite indexing to RocksDB

Replace SQLite-backed indexing and provenance storage with RocksDB-backed stores.

  Key changes:
  - add RocksDB async/database/db-manager/filesystem/key-codec layers
  - migrate index and provenance databases from SQLite to RocksDB
  - update index builder, trace reader, reorganize, view, stats, and comparator paths for
  RocksDB
  - harden transaction atomicity and rollback behavior with TransactionScope
  - add iterator status checking for prefix scans
  - harden gzip/tar indexer cache state and metadata handling
  - capture executor context in RocksDB awaitables
  - clean up failed RocksDB open paths and manager lifecycle behavior
  - vendor CPM 0.42.1 and update CI/build integration
  - refresh docs, Python bindings, and C++/Python test coverage for the new backend

  Validation:
  - full test suite passed
  - Ubuntu 22.04 Docker run passed
  - focused RocksDB/indexer regression tests passed.

Coverage Stats

24097 of 59624 branches covered (40.41%)

Branch coverage included in aggregate %.

2516 of 3144 new or added lines in 75 files covered. (80.03%)

72 existing lines in 15 files now uncovered.

20858 of 26701 relevant lines covered (78.12%)

14113.43 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

40.76

/src/dftracer/utils/utilities/reader/internal/tar_reader.cpp

#include <dftracer/utils/core/common/filesystem.h>
#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/coro/task.h>
#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
#include <dftracer/utils/utilities/reader/internal/streams/tar_byte_stream.h>
#include <dftracer/utils/utilities/reader/internal/string_line_processor.h>
#include <dftracer/utils/utilities/reader/internal/tar_reader.h>

#include <algorithm>
#include <cstring>
#include <sstream>
#include <vector>

using namespace dftracer::utils::utilities::indexer::internal::tar;

namespace dftracer::utils::utilities::reader::internal {

namespace {

std::string normalize_idx_path(const std::string &path) {
    fs::path input(path);
    if (input.filename() == ".dftindex") {
        return input.string();
    }
    if (input.parent_path().filename() == ".dftindex") {
        return input.parent_path().string();
    }
    if (input.has_extension()) {
        return (input.parent_path() / ".dftindex").string();
    }
    return (input / ".dftindex").string();
}

}  // namespace

TarReader::TarReader(const std::string &tar_gz_path_,
                     const std::string &idx_path_, std::size_t index_ckpt_size)
    : tar_gz_path(tar_gz_path_),
      index_path(normalize_idx_path(idx_path_)),
      is_open(false),
      default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),
      logical_mapping_cached(false),
      cached_total_logical_bytes(0),
      cached_total_logical_lines(0) {
    try {
        printf("Creating TAR reader for gz: %s and index: %s\n",
               tar_gz_path.c_str(), index_path.c_str());
        indexer = std::make_shared<TarIndexer>(tar_gz_path, index_path,
                                               index_ckpt_size, false);
        is_open = true;

        DFTRACER_UTILS_LOG_DEBUG(
            "Successfully created TAR reader for gz: %s and index: %s",
            tar_gz_path.c_str(), index_path.c_str());
    } catch (const std::exception &e) {
        throw std::runtime_error(
            "Failed to initialize TAR reader with indexer: " +
            std::string(e.what()));
    }
}

TarReader::TarReader(std::shared_ptr<TarIndexer> indexer_)
    : default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),
      indexer(indexer_),
      logical_mapping_cached(false),
      cached_total_logical_bytes(0),
      cached_total_logical_lines(0) {
    if (indexer == nullptr) {
        throw std::runtime_error("Invalid indexer provided");
    }
    is_open = true;
    tar_gz_path = indexer->get_tar_gz_path();
    index_path = indexer->get_index_path();
}

TarReader::~TarReader() = default;

TarReader::TarReader(TarReader &&other) noexcept
    : tar_gz_path(std::move(other.tar_gz_path)),
      index_path(std::move(other.index_path)),
      is_open(other.is_open),
      default_buffer_size(other.default_buffer_size),
      indexer(std::move(other.indexer)),
      logical_mapping_cached(other.logical_mapping_cached),
      cached_file_mapping(std::move(other.cached_file_mapping)),
      cached_total_logical_bytes(other.cached_total_logical_bytes),
      cached_total_logical_lines(other.cached_total_logical_lines) {
    other.is_open = false;
    other.logical_mapping_cached = false;
}

TarReader &TarReader::operator=(TarReader &&other) noexcept {
    if (this != &other) {
        tar_gz_path = std::move(other.tar_gz_path);
        index_path = std::move(other.index_path);
        is_open = other.is_open;
        default_buffer_size = other.default_buffer_size;
        indexer = std::move(other.indexer);
        logical_mapping_cached = other.logical_mapping_cached;
        cached_file_mapping = std::move(other.cached_file_mapping);
        cached_total_logical_bytes = other.cached_total_logical_bytes;
        cached_total_logical_lines = other.cached_total_logical_lines;

        other.is_open = false;
        other.logical_mapping_cached = false;
    }
    return *this;
}

// Metadata operations
std::size_t TarReader::get_max_bytes() const {
    build_logical_mapping();
    return cached_total_logical_bytes;
}

std::size_t TarReader::get_num_lines() const {
    build_logical_mapping();
    return cached_total_logical_lines;
}

std::string TarReader::get_format_name() const { return "TAR.GZ"; }

const std::string &TarReader::get_archive_path() const { return tar_gz_path; }

const std::string &TarReader::get_index_path() const { return index_path; }

void TarReader::set_buffer_size(std::size_t size) {
    default_buffer_size = size;
}

std::unique_ptr<ReaderStream> TarReader::stream(
    [[maybe_unused]] const StreamConfig &config) {
    // TODO: Implement TAR stream creation
    throw std::runtime_error("TarReader::stream() not yet implemented");
}

// Archive structure operations
std::vector<TarReader::TarFileInfo> TarReader::list_files() const {
    build_logical_mapping();
    return cached_file_mapping;
}

// Reader interface - operates on concatenated logical view
coro::CoroTask<std::size_t> TarReader::read_async(std::size_t start_bytes,
                                                  std::size_t end_bytes,
                                                  char *buffer,
                                                  std::size_t buffer_size) {
    co_return read_logical(start_bytes, end_bytes, buffer, buffer_size);
}

coro::CoroTask<std::size_t> TarReader::read_line_bytes_async(
    std::size_t start_bytes, std::size_t end_bytes, char *buffer,
    std::size_t buffer_size) {
    build_logical_mapping();

    if (start_bytes >= cached_total_logical_bytes || buffer_size == 0) {
        co_return 0;
    }

    std::size_t actual_end = std::min(
        end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));

    // Read a larger chunk to ensure we get complete lines
    std::size_t read_size = std::min(actual_end - start_bytes, buffer_size * 2);
    std::vector<char> temp_buffer(read_size);

    std::size_t bytes_read = read_logical(start_bytes, start_bytes + read_size,
                                          temp_buffer.data(), read_size);

    if (bytes_read == 0) {
        co_return 0;
    }

    // Find the first line start (if start_bytes is not at line beginning)
    std::size_t line_start = 0;
    if (start_bytes > 0) {
        // Look back to find the beginning of the current line
        std::size_t lookback_start =
            (start_bytes >= 512) ? start_bytes - 512 : 0;
        std::vector<char> lookback_buffer(512);
        std::size_t lookback_read = read_logical(lookback_start, start_bytes,
                                                 lookback_buffer.data(), 512);

        if (lookback_read > 0) {
            // Find the last newline in the lookback buffer
            for (std::size_t i = lookback_read; i > 0; --i) {
                if (lookback_buffer[i - 1] == '\n') {
                    // The line starts after this newline, but we need to adjust
                    // for the position difference
                    break;
                }
            }
        }
    }

    // Find complete lines within our read data
    std::size_t output_pos = 0;
    std::size_t search_pos = line_start;

    while (search_pos < bytes_read && output_pos < buffer_size) {
        // Find the next newline
        std::size_t newline_pos = search_pos;
        while (newline_pos < bytes_read && temp_buffer[newline_pos] != '\n') {
            newline_pos++;
        }

        if (newline_pos < bytes_read) {
            // Include the newline character
            newline_pos++;
            std::size_t line_len = newline_pos - search_pos;

            if (output_pos + line_len <= buffer_size) {
                std::memcpy(buffer + output_pos,
                            temp_buffer.data() + search_pos, line_len);
                output_pos += line_len;
                search_pos = newline_pos;
            } else {
                break;  // Buffer full
            }
        } else {
            // Incomplete line at the end, don't include it
            break;
        }
    }

    co_return output_pos;
}

coro::CoroTask<std::string> TarReader::read_lines_async(std::size_t start_line,
                                                        std::size_t end_line) {
    build_logical_mapping();

    if (start_line < 1 || start_line > cached_total_logical_lines) {
        co_return "";
    }

    std::size_t actual_end_line = std::min(
        end_line, static_cast<std::size_t>(cached_total_logical_lines));

    // Find which files contain the requested lines
    std::string result;
    std::size_t current_global_line = 1;

    for (const auto &file_info : cached_file_mapping) {
        // Check if this file contains any of our target lines
        if (current_global_line + file_info.estimated_lines <= start_line) {
            current_global_line += file_info.estimated_lines;
            continue;
        }
        if (current_global_line > actual_end_line) {
            break;
        }

        // Read content from this file
        std::size_t file_start_line =
            std::max(start_line, current_global_line) - current_global_line + 1;
        std::size_t file_end_line =
            std::min(actual_end_line,
                     static_cast<std::size_t>(current_global_line +
                                              file_info.estimated_lines - 1)) -
            current_global_line + 1;

        std::string file_content =
            read_file_content(file_info, 0, file_info.file_size);
        result += extract_lines_from_content(file_content, file_start_line,
                                             file_end_line);

        current_global_line += file_info.estimated_lines;
    }

    co_return result;
}

coro::CoroTask<void> TarReader::read_lines_with_processor_async(
    std::size_t start_line, std::size_t end_line, LineProcessor &processor) {
    build_logical_mapping();

    if (start_line < 1 || start_line > cached_total_logical_lines) {
        co_return;
    }

    std::size_t actual_end_line = std::min(
        end_line, static_cast<std::size_t>(cached_total_logical_lines));

    // Find which files contain the requested lines
    std::size_t current_global_line = 1;

    processor.begin(start_line, actual_end_line);

    for (const auto &file_info : cached_file_mapping) {
        // Check if this file contains any of our target lines
        if (current_global_line + file_info.estimated_lines <= start_line) {
            current_global_line += file_info.estimated_lines;
            continue;
        }
        if (current_global_line > actual_end_line) {
            break;
        }

        // Read content from this file
        std::string file_content =
            read_file_content(file_info, 0, file_info.file_size);
        process_content_lines(file_content, processor, current_global_line,
                              start_line, actual_end_line);

        current_global_line += file_info.estimated_lines;
    }

    processor.end();
}

coro::CoroTask<void> TarReader::read_line_bytes_with_processor_async(
    std::size_t start_bytes, std::size_t end_bytes, LineProcessor &processor) {
    build_logical_mapping();

    if (start_bytes >= cached_total_logical_bytes) {
        co_return;
    }

    std::size_t actual_end = std::min(
        end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));

    processor.begin(start_bytes, actual_end);

    // Find which files contain the requested byte range
    for (const auto &file_info : cached_file_mapping) {
        if (file_info.logical_end_offset <= start_bytes) {
            continue;  // This file is before our range
        }
        if (file_info.logical_start_offset >= actual_end) {
            break;     // We've passed our range
        }

        // Calculate the portion of this file we need
        std::size_t file_start =
            (start_bytes > file_info.logical_start_offset)
                ? start_bytes - file_info.logical_start_offset
                : 0;
        std::size_t file_end = std::min(
            file_info.file_size, actual_end - file_info.logical_start_offset);

        if (file_start < file_end) {
            std::string file_content =
                read_file_content(file_info, file_start, file_end - file_start);
            co_await processor.process(file_content.c_str(),
                                       file_content.length());
        }
    }

    processor.end();
}

void TarReader::reset() {
    logical_mapping_cached = false;
    cached_file_mapping.clear();
    cached_total_logical_bytes = 0;
    cached_total_logical_lines = 0;
}

bool TarReader::is_valid() const { return is_open && indexer; }

// File-level operations - NOT YET IMPLEMENTED
std::string TarReader::read_file(const std::string &filename,
                                 std::size_t start_line, std::size_t end_line) {
    (void)filename;
    (void)start_line;
    (void)end_line;  // Suppress unused warnings
    return "";       // Placeholder
}

bool TarReader::find_file(const std::string &filename,
                          TarFileInfo &file_info) const {
    build_logical_mapping();

    auto it =
        std::find_if(cached_file_mapping.begin(), cached_file_mapping.end(),
                     [&filename](const TarFileInfo &info) {
                         return info.file_name == filename;
                     });

    if (it != cached_file_mapping.end()) {
        file_info = *it;
        return true;
    }

    return false;
}

// Internal helper methods implementation moved from TarReaderImplementor
void TarReader::build_logical_mapping() const {
    if (logical_mapping_cached) {
        return;
    }

    DFTRACER_UTILS_LOG_DEBUG(
        "%s", "Building logical content mapping for TAR.GZ file");

    try {
        // Get all TAR file entries from the indexer
        auto tar_files = indexer->list_files();

        cached_file_mapping.clear();
        cached_file_mapping.reserve(tar_files.size());

        std::uint64_t logical_offset = 0;
        std::uint64_t logical_line = 1;

        for (const auto &tar_file : tar_files) {
            TarFileInfo file_info;
            file_info.file_name = tar_file.file_name;
            file_info.file_size = tar_file.file_size;
            file_info.file_mtime = tar_file.file_mtime;
            file_info.typeflag = tar_file.typeflag;
            file_info.logical_start_offset = logical_offset;
            file_info.logical_start_line = logical_line;

            // Calculate end positions
            file_info.logical_end_offset = logical_offset + tar_file.file_size;

            // Estimate lines in this file (rough approximation)
            file_info.estimated_lines =
                tar_file.file_size > 0 ? (tar_file.file_size / 50) + 1 : 1;
            file_info.logical_end_line =
                logical_line + file_info.estimated_lines - 1;

            cached_file_mapping.push_back(file_info);

            logical_offset = file_info.logical_end_offset;
            logical_line = file_info.logical_end_line + 1;
        }

        cached_total_logical_bytes = logical_offset;
        cached_total_logical_lines = logical_line - 1;
        logical_mapping_cached = true;

        DFTRACER_UTILS_LOG_DEBUG(
            "Built logical mapping: %zu files, %zu bytes, %zu lines",
            cached_file_mapping.size(), cached_total_logical_bytes,
            cached_total_logical_lines);
    } catch (const std::exception &e) {
        throw std::runtime_error("Failed to build TAR logical mapping: " +
                                 std::string(e.what()));
    }
}

std::size_t TarReader::read_logical(std::size_t start_bytes,
                                    std::size_t end_bytes, char *buffer,
                                    std::size_t buffer_size) const {
    build_logical_mapping();

    if (start_bytes >= cached_total_logical_bytes || buffer_size == 0) {
        return 0;
    }

    std::size_t actual_end = std::min(
        end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));
    [[maybe_unused]] std::size_t total_to_read = actual_end - start_bytes;
    [[maybe_unused]] std::size_t total_written = 0;

    DFTRACER_UTILS_LOG_DEBUG(
        "TAR logical read: start=%zu, end=%zu, buffer_size=%zu", start_bytes,
        actual_end, buffer_size);

    // Find which files contain the requested byte range
    std::size_t bytes_written = 0;

    for (const auto &file_info : cached_file_mapping) {
        if (file_info.logical_end_offset <= start_bytes) {
            continue;  // This file is before our range
        }
        if (file_info.logical_start_offset >= actual_end) {
            break;     // We've passed our range
        }
        if (bytes_written >= buffer_size) {
            break;     // Buffer is full
        }

        // Calculate the portion of this file we need
        std::size_t file_start =
            (start_bytes > file_info.logical_start_offset)
                ? start_bytes - file_info.logical_start_offset
                : 0;
        std::size_t file_end = std::min(
            file_info.file_size, actual_end - file_info.logical_start_offset);

        if (file_start < file_end) {
            std::size_t bytes_to_read =
                std::min(file_end - file_start, buffer_size - bytes_written);
            std::string file_content =
                read_file_content(file_info, file_start, bytes_to_read);

            std::memcpy(buffer + bytes_written, file_content.c_str(),
                        file_content.length());
            bytes_written += file_content.length();
        }
    }

    DFTRACER_UTILS_LOG_DEBUG("TAR logical read: returning %zu bytes",
                             bytes_written);
    return bytes_written;
}

// Helper method implementations
std::string TarReader::read_file_content(const TarFileInfo &file_info,
                                         std::size_t offset,
                                         std::size_t size) const {
    if (size == 0) {
        return "";
    }

    try {
        // Calculate the absolute offset in the logical stream where the file
        // data starts
        std::size_t file_data_start = file_info.logical_start_offset + offset;
        std::size_t file_data_end = file_data_start + size;

        // Ensure we don't read beyond the file boundaries
        std::size_t max_end =
            file_info.logical_start_offset + file_info.file_size;
        if (file_data_end > max_end) {
            file_data_end = max_end;
        }

        if (file_data_start >= file_data_end) {
            return "";
        }

        // Use a buffer to read the data
        std::size_t actual_size = file_data_end - file_data_start;
        std::vector<char> buffer(actual_size);

        // Create a TAR byte stream to read file content
        auto byte_stream = std::make_unique<TarByteStream>();
        byte_stream->initialize(tar_gz_path, file_data_start, file_data_end,
                                *indexer);

        std::size_t total_read = 0;
        while (total_read < actual_size) {
            std::size_t chunk_size = std::min(actual_size - total_read,
                                              static_cast<std::size_t>(8192));
            std::size_t bytes_read =
                byte_stream->read(buffer.data() + total_read, chunk_size);

            if (bytes_read == 0) {
                break;  // EOF or error
            }

            total_read += bytes_read;
        }

        return std::string(buffer.data(), total_read);

    } catch (const std::exception &e) {
        DFTRACER_UTILS_LOG_DEBUG("Error reading TAR file content: %s",
                                 e.what());
        return "";
    }
}

std::string TarReader::extract_lines_from_content(const std::string &content,
                                                  std::size_t start_line,
                                                  std::size_t end_line) const {
    if (content.empty()) {
        return "";
    }

    std::istringstream stream(content);
    std::string line;
    std::string result;
    std::size_t current_line = 1;

    while (std::getline(stream, line) && current_line <= end_line) {
        if (current_line >= start_line) {
            result += line + "\n";
        }
        current_line++;
    }

    return result;
}

void TarReader::process_content_lines(const std::string &content,
                                      LineProcessor &processor,
                                      std::size_t base_line,
                                      std::size_t start_line,
                                      std::size_t end_line) const {
    if (content.empty()) {
        return;
    }

    std::istringstream stream(content);
    std::string line;
    std::size_t current_line = base_line;

    while (std::getline(stream, line) && current_line <= end_line) {
        if (current_line >= start_line) {
            // Sync C API path — .get() intentional
            if (!processor.process(line.c_str(), line.length()).get()) {
                break;
            }
        }
        current_line++;
    }
}

}  // namespace dftracer::utils::utilities::reader::internal

1	#include <dftracer/utils/core/common/filesystem.h>
2	#include <dftracer/utils/core/common/logging.h>
3	#include <dftracer/utils/core/coro/task.h>
4	#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
5	#include <dftracer/utils/utilities/reader/internal/streams/tar_byte_stream.h>
6	#include <dftracer/utils/utilities/reader/internal/string_line_processor.h>
7	#include <dftracer/utils/utilities/reader/internal/tar_reader.h>
8
9	#include <algorithm>
10	#include <cstring>
11	#include <sstream>
12	#include <vector>
13
14	using namespace dftracer::utils::utilities::indexer::internal::tar;
15
16	namespace dftracer::utils::utilities::reader::internal {
17
18	namespace {
19
20	std::string normalize_idx_path(const std::string &path) {	2✔
21	fs::path input(path);	2!
22	if (input.filename() == ".dftindex") {	2!
23	return input.string();	2!
24	}
NEW 25	if (input.parent_path().filename() == ".dftindex") {	×
NEW 26	return input.parent_path().string();	×
27	}
NEW 28	if (input.has_extension()) {	×
NEW 29	return (input.parent_path() / ".dftindex").string();	×
30	}
NEW 31	return (input / ".dftindex").string();	×
32	}	2✔
33
34	} // namespace
35
36	TarReader::TarReader(const std::string &tar_gz_path_,	5✔
37	const std::string &idx_path_, std::size_t index_ckpt_size)	1✔
38	: tar_gz_path(tar_gz_path_),	2!
39	index_path(normalize_idx_path(idx_path_)),	2!
40	is_open(false),	2✔
41	default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),	2✔
42	logical_mapping_cached(false),	2✔
43	cached_total_logical_bytes(0),	2✔
44	cached_total_logical_lines(0) {	5!
45	try {
46	printf("Creating TAR reader for gz: %s and index: %s\n",	2!
47	tar_gz_path.c_str(), index_path.c_str());	1✔
48	indexer = std::make_shared<TarIndexer>(tar_gz_path, index_path,	3!
49	index_ckpt_size, false);	2!
50	is_open = true;	2✔
51
52	DFTRACER_UTILS_LOG_DEBUG(
53	"Successfully created TAR reader for gz: %s and index: %s",
54	tar_gz_path.c_str(), index_path.c_str());
55	} catch (const std::exception &e) {	1!
56	throw std::runtime_error(	×
57	"Failed to initialize TAR reader with indexer: " +	×
58	std::string(e.what()));	×
59	}	×
60	}	3✔
61
62	TarReader::TarReader(std::shared_ptr<TarIndexer> indexer_)	55✔
63	: default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),	22✔
64	indexer(indexer_),	22✔
65	logical_mapping_cached(false),	22✔
66	cached_total_logical_bytes(0),	22✔
67	cached_total_logical_lines(0) {	55✔
68	if (indexer == nullptr) {	22✔
69	throw std::runtime_error("Invalid indexer provided");	×
70	}
71	is_open = true;	22✔
72	tar_gz_path = indexer->get_tar_gz_path();	22!
73	index_path = indexer->get_index_path();	22!
74	}	33✔
75
76	TarReader::~TarReader() = default;	36✔
77
78	TarReader::TarReader(TarReader &&other) noexcept	×
79	: tar_gz_path(std::move(other.tar_gz_path)),	×
NEW 80	index_path(std::move(other.index_path)),	×
81	is_open(other.is_open),	×
82	default_buffer_size(other.default_buffer_size),	×
83	indexer(std::move(other.indexer)),	×
84	logical_mapping_cached(other.logical_mapping_cached),	×
85	cached_file_mapping(std::move(other.cached_file_mapping)),	×
86	cached_total_logical_bytes(other.cached_total_logical_bytes),	×
87	cached_total_logical_lines(other.cached_total_logical_lines) {	×
88	other.is_open = false;	×
89	other.logical_mapping_cached = false;	×
90	}	×
91
92	TarReader &TarReader::operator=(TarReader &&other) noexcept {	×
93	if (this != &other) {	×
94	tar_gz_path = std::move(other.tar_gz_path);	×
NEW 95	index_path = std::move(other.index_path);	×
96	is_open = other.is_open;	×
97	default_buffer_size = other.default_buffer_size;	×
98	indexer = std::move(other.indexer);	×
99	logical_mapping_cached = other.logical_mapping_cached;	×
100	cached_file_mapping = std::move(other.cached_file_mapping);	×
101	cached_total_logical_bytes = other.cached_total_logical_bytes;	×
102	cached_total_logical_lines = other.cached_total_logical_lines;	×
103
104	other.is_open = false;	×
105	other.logical_mapping_cached = false;	×
106	}
107	return *this;	×
108	}
109
110	// Metadata operations
111	std::size_t TarReader::get_max_bytes() const {	2✔
112	build_logical_mapping();	2✔
113	return cached_total_logical_bytes;	2✔
114	}
115
116	std::size_t TarReader::get_num_lines() const {	×
117	build_logical_mapping();	×
118	return cached_total_logical_lines;	×
119	}
120
121	std::string TarReader::get_format_name() const { return "TAR.GZ"; }	×
122
123	const std::string &TarReader::get_archive_path() const { return tar_gz_path; }	2✔
124
125	const std::string &TarReader::get_index_path() const { return index_path; }	2✔
126
127	void TarReader::set_buffer_size(std::size_t size) {	×
128	default_buffer_size = size;	×
129	}	×
130
131	std::unique_ptr<ReaderStream> TarReader::stream(	×
132	[[maybe_unused]] const StreamConfig &config) {
133	// TODO: Implement TAR stream creation
134	throw std::runtime_error("TarReader::stream() not yet implemented");	×
135	}
136
137	// Archive structure operations
138	std::vector<TarReader::TarFileInfo> TarReader::list_files() const {	×
139	build_logical_mapping();	×
140	return cached_file_mapping;	×
141	}
142
143	// Reader interface - operates on concatenated logical view
144	coro::CoroTask<std::size_t> TarReader::read_async(std::size_t start_bytes,	918!
145	std::size_t end_bytes,
146	char *buffer,
147	std::size_t buffer_size) {	153!
148	co_return read_logical(start_bytes, end_bytes, buffer, buffer_size);	306!
149	}	306!
150
151	coro::CoroTask<std::size_t> TarReader::read_line_bytes_async(	6!
152	std::size_t start_bytes, std::size_t end_bytes, char *buffer,
153	std::size_t buffer_size) {	1!
154	build_logical_mapping();	1!
155
156	if (start_bytes >= cached_total_logical_bytes \|\| buffer_size == 0) {	1!
157	co_return 0;	1!
158	}
159
160	std::size_t actual_end = std::min(	2!
161	end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));	1✔
162
163	// Read a larger chunk to ensure we get complete lines
164	std::size_t read_size = std::min(actual_end - start_bytes, buffer_size * 2);	1!
165	std::vector<char> temp_buffer(read_size);	1!
166
167	std::size_t bytes_read = read_logical(start_bytes, start_bytes + read_size,	2!
168	temp_buffer.data(), read_size);	1✔
169
170	if (bytes_read == 0) {	1!
171	co_return 0;	×
172	}
173
174	// Find the first line start (if start_bytes is not at line beginning)
175	std::size_t line_start = 0;	1✔
176	if (start_bytes > 0) {	1!
177	// Look back to find the beginning of the current line
178	std::size_t lookback_start =
179	(start_bytes >= 512) ? start_bytes - 512 : 0;	×
180	std::vector<char> lookback_buffer(512);	×
181	std::size_t lookback_read = read_logical(lookback_start, start_bytes,	×
182	lookback_buffer.data(), 512);
183
184	if (lookback_read > 0) {	×
185	// Find the last newline in the lookback buffer
186	for (std::size_t i = lookback_read; i > 0; --i) {	×
187	if (lookback_buffer[i - 1] == '\n') {	×
188	// The line starts after this newline, but we need to adjust
189	// for the position difference
190	break;
191	}
192	}
193	}
194	}
195
196	// Find complete lines within our read data
197	std::size_t output_pos = 0;	1✔
198	std::size_t search_pos = line_start;	1✔
199
200	while (search_pos < bytes_read && output_pos < buffer_size) {	17!
201	// Find the next newline
202	std::size_t newline_pos = search_pos;	17✔
203	while (newline_pos < bytes_read && temp_buffer[newline_pos] != '\n') {	1,036!
204	newline_pos++;	1,019✔
205	}
206
207	if (newline_pos < bytes_read) {	17!
208	// Include the newline character
209	newline_pos++;	17✔
210	std::size_t line_len = newline_pos - search_pos;	17✔
211
212	if (output_pos + line_len <= buffer_size) {	17✔
213	std::memcpy(buffer + output_pos,	32✔
214	temp_buffer.data() + search_pos, line_len);	16✔
215	output_pos += line_len;	16✔
216	search_pos = newline_pos;	16✔
217	} else {	16✔
218	break; // Buffer full	1✔
219	}
220	} else {	17✔
221	// Incomplete line at the end, don't include it
222	break;
223	}
224	}	17✔
225
226	co_return output_pos;	1!
227	}	3!
228
229	coro::CoroTask<std::string> TarReader::read_lines_async(std::size_t start_line,	96!
230	std::size_t end_line) {	16!
231	build_logical_mapping();	16!
232
233	if (start_line < 1 \|\| start_line > cached_total_logical_lines) {	16!
234	co_return "";	16!
235	}
236
237	std::size_t actual_end_line = std::min(	32!
238	end_line, static_cast<std::size_t>(cached_total_logical_lines));	16✔
239
240	// Find which files contain the requested lines
241	std::string result;	16✔
242	std::size_t current_global_line = 1;	16✔
243
244	for (const auto &file_info : cached_file_mapping) {	50!
245	// Check if this file contains any of our target lines
246	if (current_global_line + file_info.estimated_lines <= start_line) {	34✔
247	current_global_line += file_info.estimated_lines;	2✔
248	continue;	2✔
249	}
250	if (current_global_line > actual_end_line) {	32✔
251	break;	16✔
252	}
253
254	// Read content from this file
255	std::size_t file_start_line =	16✔
256	std::max(start_line, current_global_line) - current_global_line + 1;	16!
257	std::size_t file_end_line =	16✔
258	std::min(actual_end_line,	16!
259	static_cast<std::size_t>(current_global_line +	48✔
260	file_info.estimated_lines - 1)) -	32✔
261	current_global_line + 1;	32✔
262
263	std::string file_content =	16✔
264	read_file_content(file_info, 0, file_info.file_size);	16!
265	result += extract_lines_from_content(file_content, file_start_line,	32!
266	file_end_line);	16✔
267
268	current_global_line += file_info.estimated_lines;	16✔
269	}	34✔
270
271	co_return result;	16!
272	}	48!
273
274	coro::CoroTask<void> TarReader::read_lines_with_processor_async(	×
275	std::size_t start_line, std::size_t end_line, LineProcessor &processor) {	×
276	build_logical_mapping();	×
277
278	if (start_line < 1 \|\| start_line > cached_total_logical_lines) {	×
279	co_return;
280	}
281
282	std::size_t actual_end_line = std::min(	×
283	end_line, static_cast<std::size_t>(cached_total_logical_lines));
284
285	// Find which files contain the requested lines
286	std::size_t current_global_line = 1;
287
288	processor.begin(start_line, actual_end_line);	×
289
290	for (const auto &file_info : cached_file_mapping) {	×
291	// Check if this file contains any of our target lines
292	if (current_global_line + file_info.estimated_lines <= start_line) {	×
293	current_global_line += file_info.estimated_lines;
294	continue;
295	}
296	if (current_global_line > actual_end_line) {	×
297	break;
298	}
299
300	// Read content from this file
301	std::string file_content =
302	read_file_content(file_info, 0, file_info.file_size);	×
303	process_content_lines(file_content, processor, current_global_line,	×
304	start_line, actual_end_line);
305
306	current_global_line += file_info.estimated_lines;
307	}	×
308
309	processor.end();	×
310	}	×
311
312	coro::CoroTask<void> TarReader::read_line_bytes_with_processor_async(	×
313	std::size_t start_bytes, std::size_t end_bytes, LineProcessor &processor) {	×
314	build_logical_mapping();	×
315
316	if (start_bytes >= cached_total_logical_bytes) {	×
317	co_return;
318	}
319
320	std::size_t actual_end = std::min(	×
321	end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));
322
323	processor.begin(start_bytes, actual_end);	×
324
325	// Find which files contain the requested byte range
326	for (const auto &file_info : cached_file_mapping) {	×
327	if (file_info.logical_end_offset <= start_bytes) {	×
328	continue; // This file is before our range
329	}
330	if (file_info.logical_start_offset >= actual_end) {	×
331	break; // We've passed our range
332	}
333
334	// Calculate the portion of this file we need
335	std::size_t file_start =
336	(start_bytes > file_info.logical_start_offset)	×
337	? start_bytes - file_info.logical_start_offset
338	: 0;
339	std::size_t file_end = std::min(	×
340	file_info.file_size, actual_end - file_info.logical_start_offset);
341
342	if (file_start < file_end) {	×
343	std::string file_content =
344	read_file_content(file_info, file_start, file_end - file_start);	×
345	co_await processor.process(file_content.c_str(),	×
346	file_content.length());
347	}	×
348	}	×
349
350	processor.end();	×
351	}	×
352
353	void TarReader::reset() {	×
354	logical_mapping_cached = false;	×
355	cached_file_mapping.clear();	×
356	cached_total_logical_bytes = 0;	×
357	cached_total_logical_lines = 0;	×
358	}	×
359
360	bool TarReader::is_valid() const { return is_open && indexer; }	6!
361
362	// File-level operations - NOT YET IMPLEMENTED
363	std::string TarReader::read_file(const std::string &filename,	×
364	std::size_t start_line, std::size_t end_line) {
365	(void)filename;
366	(void)start_line;
367	(void)end_line; // Suppress unused warnings
368	return ""; // Placeholder	×
369	}
370
371	bool TarReader::find_file(const std::string &filename,	×
372	TarFileInfo &file_info) const {
373	build_logical_mapping();	×
374
375	auto it =
376	std::find_if(cached_file_mapping.begin(), cached_file_mapping.end(),	×
377	[&filename](const TarFileInfo &info) {	×
378	return info.file_name == filename;	×
379	});
380
381	if (it != cached_file_mapping.end()) {	×
382	file_info = *it;	×
383	return true;	×
384	}
385
386	return false;	×
387	}
388
389	// Internal helper methods implementation moved from TarReaderImplementor
390	void TarReader::build_logical_mapping() const {	344✔
391	if (logical_mapping_cached) {	344✔
392	return;	326✔
393	}
394
395	DFTRACER_UTILS_LOG_DEBUG(
396	"%s", "Building logical content mapping for TAR.GZ file");
397
398	try {
399	// Get all TAR file entries from the indexer
400	auto tar_files = indexer->list_files();	18!
401
402	cached_file_mapping.clear();	18✔
403	cached_file_mapping.reserve(tar_files.size());	18!
404
405	std::uint64_t logical_offset = 0;	18✔
406	std::uint64_t logical_line = 1;	18✔
407
408	for (const auto &tar_file : tar_files) {	90✔
409	TarFileInfo file_info;	72✔
410	file_info.file_name = tar_file.file_name;	72!
411	file_info.file_size = tar_file.file_size;	72✔
412	file_info.file_mtime = tar_file.file_mtime;	72✔
413	file_info.typeflag = tar_file.typeflag;	72✔
414	file_info.logical_start_offset = logical_offset;	72✔
415	file_info.logical_start_line = logical_line;	72✔
416
417	// Calculate end positions
418	file_info.logical_end_offset = logical_offset + tar_file.file_size;	72✔
419
420	// Estimate lines in this file (rough approximation)
421	file_info.estimated_lines =	72✔
422	tar_file.file_size > 0 ? (tar_file.file_size / 50) + 1 : 1;	72!
423	file_info.logical_end_line =	72✔
424	logical_line + file_info.estimated_lines - 1;	72✔
425
426	cached_file_mapping.push_back(file_info);	72!
427
428	logical_offset = file_info.logical_end_offset;	72✔
429	logical_line = file_info.logical_end_line + 1;	72✔
430	}	72✔
431
432	cached_total_logical_bytes = logical_offset;	18✔
433	cached_total_logical_lines = logical_line - 1;	18✔
434	logical_mapping_cached = true;	18✔
435
436	DFTRACER_UTILS_LOG_DEBUG(
437	"Built logical mapping: %zu files, %zu bytes, %zu lines",
438	cached_file_mapping.size(), cached_total_logical_bytes,
439	cached_total_logical_lines);
440	} catch (const std::exception &e) {	18!
441	throw std::runtime_error("Failed to build TAR logical mapping: " +	×
442	std::string(e.what()));	×
443	}	×
444	}	172✔
445
446	std::size_t TarReader::read_logical(std::size_t start_bytes,	308✔
447	std::size_t end_bytes, char *buffer,
448	std::size_t buffer_size) const {
449	build_logical_mapping();	308✔
450
451	if (start_bytes >= cached_total_logical_bytes \|\| buffer_size == 0) {	308✔
452	return 0;	4✔
453	}
454
455	std::size_t actual_end = std::min(	304✔
456	end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));	304✔
457	[[maybe_unused]] std::size_t total_to_read = actual_end - start_bytes;	304✔
458	[[maybe_unused]] std::size_t total_written = 0;	304✔
459
460	DFTRACER_UTILS_LOG_DEBUG(
461	"TAR logical read: start=%zu, end=%zu, buffer_size=%zu", start_bytes,
462	actual_end, buffer_size);
463
464	// Find which files contain the requested byte range
465	std::size_t bytes_written = 0;	304✔
466
467	for (const auto &file_info : cached_file_mapping) {	728✔
468	if (file_info.logical_end_offset <= start_bytes) {	722✔
469	continue; // This file is before our range	98✔
470	}
471	if (file_info.logical_start_offset >= actual_end) {	624✔
472	break; // We've passed our range	294✔
473	}
474	if (bytes_written >= buffer_size) {	330✔
475	break; // Buffer is full	4✔
476	}
477
478	// Calculate the portion of this file we need
479	std::size_t file_start =	326✔
480	(start_bytes > file_info.logical_start_offset)	326✔
481	? start_bytes - file_info.logical_start_offset	309✔
482	: 0;
483	std::size_t file_end = std::min(	326✔
484	file_info.file_size, actual_end - file_info.logical_start_offset);	326✔
485
486	if (file_start < file_end) {	326✔
487	std::size_t bytes_to_read =	163✔
488	std::min(file_end - file_start, buffer_size - bytes_written);	326✔
489	std::string file_content =
490	read_file_content(file_info, file_start, bytes_to_read);	326!
491
492	std::memcpy(buffer + bytes_written, file_content.c_str(),	489✔
493	file_content.length());	163✔
494	bytes_written += file_content.length();	326✔
495	}	326✔
496	}
497
498	DFTRACER_UTILS_LOG_DEBUG("TAR logical read: returning %zu bytes",
499	bytes_written);
500	return bytes_written;	304✔
501	}	154✔
502
503	// Helper method implementations
504	std::string TarReader::read_file_content(const TarFileInfo &file_info,	358✔
505	std::size_t offset,
506	std::size_t size) const {
507	if (size == 0) {	358✔
508	return "";	×
509	}
510
511	try {
512	// Calculate the absolute offset in the logical stream where the file
513	// data starts
514	std::size_t file_data_start = file_info.logical_start_offset + offset;	358✔
515	std::size_t file_data_end = file_data_start + size;	358✔
516
517	// Ensure we don't read beyond the file boundaries
518	std::size_t max_end =	358✔
519	file_info.logical_start_offset + file_info.file_size;	358✔
520	if (file_data_end > max_end) {	358✔
521	file_data_end = max_end;	×
522	}
523
524	if (file_data_start >= file_data_end) {	358✔
525	return "";	×
526	}
527
528	// Use a buffer to read the data
529	std::size_t actual_size = file_data_end - file_data_start;	358✔
530	std::vector<char> buffer(actual_size);	358!
531
532	// Create a TAR byte stream to read file content
533	auto byte_stream = std::make_unique<TarByteStream>();	358!
534	byte_stream->initialize(tar_gz_path, file_data_start, file_data_end,	716!
535	*indexer);	358✔
536
537	std::size_t total_read = 0;	358✔
538	while (total_read < actual_size) {	782✔
539	std::size_t chunk_size = std::min(actual_size - total_read,	848!
540	static_cast<std::size_t>(8192));	424✔
541	std::size_t bytes_read =	212✔
542	byte_stream->read(buffer.data() + total_read, chunk_size);	424!
543
544	if (bytes_read == 0) {	424✔
545	break; // EOF or error	×
546	}
547
548	total_read += bytes_read;	424✔
549	}
550
551	return std::string(buffer.data(), total_read);	358!
552
553	} catch (const std::exception &e) {	358!
554	DFTRACER_UTILS_LOG_DEBUG("Error reading TAR file content: %s",
555	e.what());
556	return "";	×
557	}	×
558	}	179✔
559
560	std::string TarReader::extract_lines_from_content(const std::string &content,	32✔
561	std::size_t start_line,
562	std::size_t end_line) const {
563	if (content.empty()) {	32!
564	return "";	×
565	}
566
567	std::istringstream stream(content);	32!
568	std::string line;	32✔
569	std::string result;	32✔
570	std::size_t current_line = 1;	32✔
571
572	while (std::getline(stream, line) && current_line <= end_line) {	650!
573	if (current_line >= start_line) {	618✔
574	result += line + "\n";	32!
575	}	16✔
576	current_line++;	618✔
577	}
578
579	return result;	32✔
580	}	32!
581
582	void TarReader::process_content_lines(const std::string &content,	×
583	LineProcessor &processor,
584	std::size_t base_line,
585	std::size_t start_line,
586	std::size_t end_line) const {
587	if (content.empty()) {	×
588	return;	×
589	}
590
591	std::istringstream stream(content);	×
592	std::string line;	×
593	std::size_t current_line = base_line;	×
594
595	while (std::getline(stream, line) && current_line <= end_line) {	×
596	if (current_line >= start_line) {	×
597	// Sync C API path — .get() intentional
598	if (!processor.process(line.c_str(), line.length()).get()) {	×
599	break;	×
600	}
601	}
602	current_line++;	×
603	}
604	}	×
605
606	} // namespace dftracer::utils::utilities::reader::internal

llnl / dftracer-utils / 24057299873

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous