29043022308

Committed 09 Jul 2026 07:04PM UTC coverage: 52.505% (-0.07%) from 52.577%

Build # 29043022308

Build Type

push

github

Committed by

hariharan-devarajan

Commit Message

feat(wheel): make wheel leaner

Coverage Stats

37438 of 92666 branches covered (40.4%)

Branch coverage included in aggregate %.

33474 of 42392 relevant lines covered (78.96%)

20907.14 hits per line

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

47.48

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

#include <dftracer/utils/core/coro/task.h>
#include <dftracer/utils/core/utils/timer.h>
#include <dftracer/utils/utilities/indexer/internal/indexer.h>
#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
#include <dftracer/utils/utilities/reader/error.h>
#include <dftracer/utils/utilities/reader/internal/gzip_reader.h>
#include <dftracer/utils/utilities/reader/internal/stream_config.h>
#include <dftracer/utils/utilities/reader/internal/streams/gzip_byte_stream.h>
#include <dftracer/utils/utilities/reader/internal/streams/gzip_line_byte_stream.h>
#include <dftracer/utils/utilities/reader/internal/streams/line_stream.h>
#include <dftracer/utils/utilities/reader/internal/streams/multi_line_stream.h>
#include <dftracer/utils/utilities/reader/internal/string_line_processor.h>

#include <cinttypes>
#include <cstdio>
#include <cstring>
#include <limits>
#include <string_view>

static void validate_parameters(
    const char *buffer, std::size_t buffer_size, std::size_t start_bytes,
    std::size_t end_bytes,
    std::size_t max_bytes = std::numeric_limits<std::size_t>::max()) {
    if (!buffer || buffer_size == 0) {
        throw dftracer::utils::utilities::reader::ReaderError(
            dftracer::utils::utilities::reader::ReaderError::INVALID_ARGUMENT,
            "Invalid buffer parameters");
    }
    if (start_bytes >= end_bytes) {
        throw dftracer::utils::utilities::reader::ReaderError(
            dftracer::utils::utilities::reader::ReaderError::INVALID_ARGUMENT,
            "start_bytes must be less than end_bytes");
    }
    if (max_bytes != SIZE_MAX) {
        if (end_bytes > max_bytes) {
            throw dftracer::utils::utilities::reader::ReaderError(
                dftracer::utils::utilities::reader::ReaderError::
                    INVALID_ARGUMENT,
                "end_bytes exceeds maximum available bytes");
        }
        if (start_bytes > max_bytes) {
            throw dftracer::utils::utilities::reader::ReaderError(
                dftracer::utils::utilities::reader::ReaderError::
                    INVALID_ARGUMENT,
                "start_bytes exceeds maximum available bytes");
        }
    }
}

static void check_reader_state(bool is_open, const void *indexer) {
    if (!is_open || !indexer) {
        throw dftracer::utils::utilities::reader::ReaderError(
            dftracer::utils::utilities::reader::ReaderError::
                INITIALIZATION_ERROR,
            "Reader is not open");
    }
}

static constexpr std::size_t DEFAULT_READER_BUFFER_SIZE = 1 * 1024 * 1024;

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

GzipReader::GzipReader(const std::string &gz_path_,
                       const std::string &idx_path_,
                       std::size_t index_ckpt_size)
    : gz_path(gz_path_),
      index_path(idx_path_),
      is_open(false),
      default_buffer_size(DEFAULT_READER_BUFFER_SIZE),
      indexer(nullptr) {
    try {
        indexer = dftracer::utils::utilities::indexer::internal::
            IndexerFactory::create(gz_path, index_path, index_ckpt_size, false);
        is_open = true;

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

GzipReader::GzipReader(
    std::shared_ptr<dftracer::utils::utilities::indexer::internal::Indexer>
        indexer_)
    : default_buffer_size(DEFAULT_READER_BUFFER_SIZE),
      indexer(std::move(indexer_)) {
    if (!indexer) {
        throw ReaderError(ReaderError::INITIALIZATION_ERROR,
                          "Invalid indexer provided");
    }
    is_open = true;
    gz_path = indexer->get_archive_path();
    index_path = indexer->get_index_path();
}

GzipReader::~GzipReader() {
    DFTRACER_UTILS_LOG_DEBUG("Destroying GZIP reader for gz: %s and index: %s",
                             gz_path.c_str(), index_path.c_str());
    reset();
    is_open = false;
}

GzipReader::GzipReader(GzipReader &&other) noexcept
    : gz_path(std::move(other.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)) {
    other.is_open = false;
}

GzipReader &GzipReader::operator=(GzipReader &&other) noexcept {
    if (this != &other) {
        gz_path = std::move(other.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);
        other.is_open = false;
    }
    return *this;
}

std::size_t GzipReader::get_max_bytes() const {
    check_reader_state(is_open, indexer.get());
    std::size_t max_bytes =
        static_cast<std::size_t>(indexer.get()->get_max_bytes());
    DFTRACER_UTILS_LOG_DEBUG("Maximum bytes available: %zu", max_bytes);
    return max_bytes;
}

std::size_t GzipReader::get_num_lines() const {
    check_reader_state(is_open, indexer.get());
    std::size_t num_lines = static_cast<std::size_t>(indexer->get_num_lines());
    DFTRACER_UTILS_LOG_DEBUG("Total lines available: %zu", num_lines);
    return num_lines;
}

const std::string &GzipReader::get_archive_path() const { return gz_path; }

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

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

void GzipReader::reset() {
    check_reader_state(is_open, indexer.get());
    stream_cache_.clear();
}

coro::CoroTask<std::size_t> GzipReader::read_async(std::size_t start_bytes,
                                                   std::size_t end_bytes,
                                                   char *buffer,
                                                   std::size_t buffer_size) {
    check_reader_state(is_open, indexer.get());
    validate_parameters(buffer, buffer_size, start_bytes, end_bytes,
                        indexer.get()->get_max_bytes());

    DFTRACER_UTILS_LOG_DEBUG(
        "GzipReader::read - request: start_bytes=%zu, end_bytes=%zu, "
        "buffer_size=%zu",
        start_bytes, end_bytes, buffer_size);

    // Check if we can reuse cached stream
    if (!stream_cache_.can_continue(StreamType::BYTES, gz_path, start_bytes,
                                    end_bytes)) {
        DFTRACER_UTILS_LOG_DEBUG("%s",
                                 "GzipReader::read - creating new byte stream");
        auto new_stream = stream(StreamConfig()
                                     .stream_type(StreamType::BYTES)
                                     .range_type(RangeType::BYTE_RANGE)
                                     .from(start_bytes)
                                     .to(end_bytes));
        stream_cache_.update(std::move(new_stream), StreamType::BYTES, gz_path,
                             start_bytes, end_bytes);
    } else {
        DFTRACER_UTILS_LOG_DEBUG(
            "%s", "GzipReader::read - reusing cached byte stream");
    }

    std::size_t result =
        co_await stream_cache_.get()->read_async(buffer, buffer_size);
    DFTRACER_UTILS_LOG_DEBUG("GzipReader::read - returned %zu bytes", result);

    // Update position for next potential read
    stream_cache_.update_position(start_bytes + result);

    co_return result;
}

coro::CoroTask<std::size_t> GzipReader::read_line_bytes_async(
    std::size_t start_bytes, std::size_t end_bytes, char *buffer,
    std::size_t buffer_size) {
    check_reader_state(is_open, indexer.get());

    if (end_bytes > indexer.get()->get_max_bytes()) {
        end_bytes = indexer.get()->get_max_bytes();
    }

    validate_parameters(buffer, buffer_size, start_bytes, end_bytes,
                        indexer.get()->get_max_bytes());

    // Check if we can reuse cached stream
    if (!stream_cache_.can_continue(StreamType::MULTI_LINES_BYTES, gz_path,
                                    start_bytes, end_bytes)) {
        auto new_stream = stream(StreamConfig()
                                     .stream_type(StreamType::MULTI_LINES_BYTES)
                                     .range_type(RangeType::BYTE_RANGE)
                                     .from(start_bytes)
                                     .to(end_bytes));
        stream_cache_.update(std::move(new_stream),
                             StreamType::MULTI_LINES_BYTES, gz_path,
                             start_bytes, end_bytes);
    }

    std::size_t result =
        co_await stream_cache_.get()->read_async(buffer, buffer_size);

    // Update position for next potential read
    stream_cache_.update_position(start_bytes + result);

    co_return result;
}

coro::CoroTask<std::string> GzipReader::read_lines_async(std::size_t start_line,
                                                         std::size_t end_line) {
    check_reader_state(is_open, indexer.get());

    if (start_line == 0 || end_line == 0) {
        throw ReaderError(ReaderError::INVALID_ARGUMENT,
                          "Line numbers must be 1-based (start from 1)");
    }

    if (start_line > end_line) {
        throw ReaderError(ReaderError::INVALID_ARGUMENT,
                          "Start line must be <= end line");
    }

    std::size_t total_lines = indexer.get()->get_num_lines();
    if (start_line > total_lines || end_line > total_lines) {
        throw ReaderError(ReaderError::INVALID_ARGUMENT,
                          "Line numbers exceed total lines in file (" +
                              std::to_string(total_lines) + ")");
    }

    // Check if we can reuse cached stream
    if (!stream_cache_.can_continue(StreamType::MULTI_LINES, gz_path,
                                    start_line, end_line)) {
        auto new_stream = stream(StreamConfig()
                                     .stream_type(StreamType::MULTI_LINES)
                                     .range_type(RangeType::LINE_RANGE)
                                     .from(start_line)
                                     .to(end_line));
        stream_cache_.update(std::move(new_stream), StreamType::MULTI_LINES,
                             gz_path, start_line, end_line);
    }

    std::string result;
    // Pre-allocate to avoid reallocations (like old StringLineProcessor)
    std::size_t estimated_lines = end_line - start_line + 1;
    result.reserve(estimated_lines * 100);  // Estimate ~100 bytes per line

    std::vector<char> buffer(default_buffer_size);

    while (!stream_cache_.get()->done()) {
        std::size_t bytes_read = co_await stream_cache_.get()->read_async(
            buffer.data(), buffer.size());
        if (bytes_read == 0) break;

        result.append(buffer.data(), bytes_read);
    }

    co_return result;
}

coro::CoroTask<void> GzipReader::read_lines_with_processor_async(
    std::size_t start_line, std::size_t end_line, LineProcessor &processor) {
    check_reader_state(is_open, indexer.get());

    if (start_line == 0 || end_line == 0) {
        throw ReaderError(ReaderError::INVALID_ARGUMENT,
                          "Line numbers must be 1-based (start from 1)");
    }

    if (start_line > end_line) {
        throw ReaderError(ReaderError::INVALID_ARGUMENT,
                          "Start line must be <= end line");
    }

    std::size_t total_lines = indexer.get()->get_num_lines();
    if (start_line > total_lines || end_line > total_lines) {
        throw ReaderError(ReaderError::INVALID_ARGUMENT,
                          "Line numbers exceed total lines in file (" +
                              std::to_string(total_lines) + ")");
    }

    processor.begin(start_line, end_line);

    // Create a LineStream that returns one line at a time
    auto line_stream = stream(StreamConfig()
                                  .stream_type(StreamType::LINE)
                                  .range_type(RangeType::LINE_RANGE)
                                  .from(start_line)
                                  .to(end_line));

    std::vector<char> buffer(default_buffer_size);

    while (!line_stream->done()) {
        std::size_t bytes_read =
            co_await line_stream->read_async(buffer.data(), buffer.size());
        if (bytes_read == 0) break;

        // LineStream returns one complete line with \n
        // Processor expects line without \n
        std::size_t line_length = bytes_read;
        if (line_length > 0 && buffer[line_length - 1] == '\n') {
            line_length--;
        }

        if (!co_await processor.process(buffer.data(), line_length)) {
            processor.end();
            co_return;
        }
    }

    processor.end();
}

coro::CoroTask<void> GzipReader::read_line_bytes_with_processor_async(
    std::size_t start_bytes, std::size_t end_bytes, LineProcessor &processor) {
    check_reader_state(is_open, indexer.get());

    if (end_bytes > indexer.get()->get_max_bytes()) {
        end_bytes = indexer.get()->get_max_bytes();
    }

    if (start_bytes >= end_bytes) {
        co_return;
    }

    processor.begin(start_bytes, end_bytes);

    auto lines_stream = stream(StreamConfig()
                                   .stream_type(StreamType::LINE_BYTES)
                                   .range_type(RangeType::BYTE_RANGE)
                                   .from(start_bytes)
                                   .to(end_bytes));

    std::vector<char> buffer(default_buffer_size);

    while (!lines_stream->done()) {
        std::size_t bytes_read =
            co_await lines_stream->read_async(buffer.data(), buffer.size());
        if (bytes_read == 0) break;
        co_await processor.process(buffer.data(), bytes_read);
    }

    processor.end();
}

bool GzipReader::is_valid() const { return is_open && indexer.get(); }

std::string GzipReader::get_format_name() const { return "GZIP"; }

std::unique_ptr<ReaderStream> GzipReader::stream(const StreamConfig &config) {
    check_reader_state(is_open, indexer.get());

    // Extract config parameters
    StreamType stream_type = config.stream_type();
    RangeType range_type = config.range_type();
    std::size_t start = config.start();
    std::size_t end = config.end();
    std::size_t buffer_size = config.buffer_size();
    bool extend_to_line_boundary = config.extend_to_line_boundary();

    // Convert line range to byte range if needed
    std::size_t start_bytes = start;
    std::size_t end_bytes = end;
    std::size_t actual_start_line =
        1;  // Track what line number start_bytes corresponds to

    if (range_type == RangeType::LINE_RANGE) {
        // Convert line numbers to byte offsets using checkpoints
        if (start == 0 || end == 0) {
            throw ReaderError(ReaderError::INVALID_ARGUMENT,
                              "Line numbers must be 1-based (start from 1)");
        }
        if (start > end) {
            throw ReaderError(ReaderError::INVALID_ARGUMENT,
                              "Start line must be <= end line");
        }

        std::size_t total_lines = indexer->get_num_lines();
        if (start > total_lines || end > total_lines) {
            throw ReaderError(ReaderError::INVALID_ARGUMENT,
                              "Line numbers exceed total lines in file (" +
                                  std::to_string(total_lines) + ")");
        }

        // Get checkpoints for the line range
        std::vector<
            dftracer::utils::utilities::indexer::internal::IndexerCheckpoint>
            checkpoints = indexer->get_checkpoints_for_line_range(start, end);

        DFTRACER_UTILS_LOG_DEBUG("Line range %zu-%zu: found %zu checkpoints",
                                 start, end, checkpoints.size());

        if (checkpoints.empty()) {
            // No checkpoints, read from beginning
            start_bytes = 0;
            end_bytes = indexer->get_max_bytes();
            actual_start_line = 1;
            DFTRACER_UTILS_LOG_DEBUG(
                "No checkpoints found, using full file: start_bytes=%zu, "
                "end_bytes=%zu, max_bytes=%" PRIu64,
                start_bytes, end_bytes, indexer->get_max_bytes());
        } else {
            // Use checkpoint to determine byte range.
            //
            // Checkpoint uc_offset values fall at deflate block boundaries
            // which may land in the middle of a text line.  When we start
            // decompressing from such a mid-line position the first "line"
            // seen by MultiLineStream is a partial fragment.  If
            // actual_start_line == start_line the fragment is emitted as
            // the requested first line, producing wrong content.
            //
            // To avoid this we choose a checkpoint whose last_line_num is
            // strictly less than (start - 1), guaranteeing
            // actual_start_line < start.  MultiLineStream then filters
            // out the (potentially partial) early lines before reaching
            // the requested range.
            auto all_checkpoints = indexer->get_checkpoints();
            bool found_start = false;

            // Walk checkpoints from the end to find the latest one whose
            // line range ends before (start - 1).
            for (auto it = all_checkpoints.rbegin();
                 it != all_checkpoints.rend(); ++it) {
                if (it->last_line_num < start - 1) {
                    start_bytes = it->uc_offset;
                    actual_start_line = it->last_line_num + 1;
                    found_start = true;
                    break;
                }
            }

            if (!found_start) {
                // No suitable checkpoint found -- start from beginning
                start_bytes = 0;
                actual_start_line = 1;
            }

            const auto &last_checkpoint = checkpoints.back();
            end_bytes = last_checkpoint.uc_offset + last_checkpoint.uc_size;

            DFTRACER_UTILS_LOG_DEBUG(
                "Using checkpoints: matched_first_idx=%" PRIu64
                " "
                "(first_line=%" PRIu64 ", last_line=%" PRIu64
                "), "
                "end_checkpoint_idx=%" PRIu64 " (first_line=%" PRIu64
                ", last_line=%" PRIu64
                "), "
                "byte_range=%zu-%zu, actual_start_line=%zu",
                checkpoints[0].checkpoint_idx, checkpoints[0].first_line_num,
                checkpoints[0].last_line_num, last_checkpoint.checkpoint_idx,
                last_checkpoint.first_line_num, last_checkpoint.last_line_num,
                start_bytes, end_bytes, actual_start_line);
        }
    }

    // Create appropriate stream type
    switch (stream_type) {
        case StreamType::BYTES: {
            auto byte_stream = std::make_unique<GzipByteStream>(buffer_size);
            byte_stream->initialize(gz_path, start_bytes, end_bytes, *indexer);
            return byte_stream;
        }
        case StreamType::LINE_BYTES: {
            // Single line-aligned bytes at a time
            auto line_byte_stream =
                std::make_unique<GzipLineByteStream>(buffer_size);
            line_byte_stream->set_extend_to_line_boundary(
                extend_to_line_boundary);
            line_byte_stream->initialize(gz_path, start_bytes, end_bytes,
                                         *indexer);

            // Wrap with LineStream to return one line-aligned chunk at a time
            if (range_type == RangeType::LINE_RANGE) {
                return std::make_unique<LineStream>(
                    std::move(line_byte_stream), start, end, actual_start_line);
            } else {
                return std::make_unique<LineStream>(
                    std::move(line_byte_stream));
            }
        }
        case StreamType::MULTI_LINES_BYTES: {
            // Multiple line-aligned bytes per read
            auto line_byte_stream =
                std::make_unique<GzipLineByteStream>(buffer_size);
            line_byte_stream->set_extend_to_line_boundary(
                extend_to_line_boundary);
            line_byte_stream->initialize(gz_path, start_bytes, end_bytes,
                                         *indexer);
            return line_byte_stream;
        }
        case StreamType::LINE: {
            // Single parsed line per read
            auto line_byte_stream =
                std::make_unique<GzipLineByteStream>(buffer_size);
            line_byte_stream->set_extend_to_line_boundary(
                extend_to_line_boundary);
            line_byte_stream->initialize(gz_path, start_bytes, end_bytes,
                                         *indexer);

            if (range_type == RangeType::LINE_RANGE) {
                return std::make_unique<LineStream>(
                    std::move(line_byte_stream), start, end, actual_start_line);
            } else {
                return std::make_unique<LineStream>(
                    std::move(line_byte_stream));
            }
        }
        case StreamType::MULTI_LINES: {
            // Multiple parsed lines per read
            auto line_byte_stream =
                std::make_unique<GzipLineByteStream>(buffer_size);
            line_byte_stream->set_extend_to_line_boundary(
                extend_to_line_boundary);
            line_byte_stream->initialize(gz_path, start_bytes, end_bytes,
                                         *indexer);

            if (range_type == RangeType::LINE_RANGE) {
                return std::make_unique<MultiLineStream>(
                    std::move(line_byte_stream), start, end, actual_start_line);
            } else {
                return std::make_unique<MultiLineStream>(
                    std::move(line_byte_stream));
            }
        }
        default:
            throw ReaderError(ReaderError::INVALID_ARGUMENT,
                              "Invalid stream type");
    }
}

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

1	#include <dftracer/utils/core/coro/task.h>
2	#include <dftracer/utils/core/utils/timer.h>
3	#include <dftracer/utils/utilities/indexer/internal/indexer.h>
4	#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
5	#include <dftracer/utils/utilities/reader/error.h>
6	#include <dftracer/utils/utilities/reader/internal/gzip_reader.h>
7	#include <dftracer/utils/utilities/reader/internal/stream_config.h>
8	#include <dftracer/utils/utilities/reader/internal/streams/gzip_byte_stream.h>
9	#include <dftracer/utils/utilities/reader/internal/streams/gzip_line_byte_stream.h>
10	#include <dftracer/utils/utilities/reader/internal/streams/line_stream.h>
11	#include <dftracer/utils/utilities/reader/internal/streams/multi_line_stream.h>
12	#include <dftracer/utils/utilities/reader/internal/string_line_processor.h>
13
14	#include <cinttypes>
15	#include <cstdio>
16	#include <cstring>
17	#include <limits>
18	#include <string_view>
19
20	static void validate_parameters(	154,666✔
21	const char *buffer, std::size_t buffer_size, std::size_t start_bytes,
22	std::size_t end_bytes,
23	std::size_t max_bytes = std::numeric_limits<std::size_t>::max()) {
24	if (!buffer \|\| buffer_size == 0) {	154,666!
25	throw dftracer::utils::utilities::reader::ReaderError(	12!
26	dftracer::utils::utilities::reader::ReaderError::INVALID_ARGUMENT,
27	"Invalid buffer parameters");	×
28	}
29	if (start_bytes >= end_bytes) {	154,666✔
30	throw dftracer::utils::utilities::reader::ReaderError(	20!
31	dftracer::utils::utilities::reader::ReaderError::INVALID_ARGUMENT,
32	"start_bytes must be less than end_bytes");	30!
33	}
34	if (max_bytes != SIZE_MAX) {	154,646✔
35	if (end_bytes > max_bytes) {	154,646✔
36	throw dftracer::utils::utilities::reader::ReaderError(	4!
37	dftracer::utils::utilities::reader::ReaderError::
38	INVALID_ARGUMENT,
39	"end_bytes exceeds maximum available bytes");	6!
40	}
41	if (start_bytes > max_bytes) {	154,642✔
42	throw dftracer::utils::utilities::reader::ReaderError(	×
43	dftracer::utils::utilities::reader::ReaderError::
44	INVALID_ARGUMENT,
45	"start_bytes exceeds maximum available bytes");	×
46	}
47	}	77,319✔
48	}	154,654✔
49
50	static void check_reader_state(bool is_open, const void *indexer) {	160,248✔
51	if (!is_open \|\| !indexer) {	160,248!
52	throw dftracer::utils::utilities::reader::ReaderError(	×
53	dftracer::utils::utilities::reader::ReaderError::
54	INITIALIZATION_ERROR,
55	"Reader is not open");	×
56	}
57	}	160,249✔
58
59	static constexpr std::size_t DEFAULT_READER_BUFFER_SIZE = 1 * 1024 * 1024;
60
61	namespace dftracer::utils::utilities::reader::internal {
62
63	GzipReader::GzipReader(const std::string &gz_path_,	1,905!
64	const std::string &idx_path_,
65	std::size_t index_ckpt_size)	477✔
66	: gz_path(gz_path_),	953!
67	index_path(idx_path_),	952!
68	is_open(false),	953✔
69	default_buffer_size(DEFAULT_READER_BUFFER_SIZE),	953✔
70	indexer(nullptr) {	2,383!
71	try {
72	indexer = dftracer::utils::utilities::indexer::internal::	475✔
73	IndexerFactory::create(gz_path, index_path, index_ckpt_size, false);	952✔
74	is_open = true;	940✔
75
76	DFTRACER_UTILS_LOG_DEBUG(	940!
77	"Successfully created GZIP reader for gz: %s and index: %s",
78	gz_path.c_str(), index_path.c_str());
79	} catch (const std::exception &e) {	481!
80	throw ReaderError(ReaderError::INITIALIZATION_ERROR,	18!
81	"Failed to initialize reader with indexer: " +	18!
82	std::string(e.what()));	18!
83	}	12!
84	}	1,466✔
85
86	GzipReader::GzipReader(	639!
87	std::shared_ptr<dftracer::utils::utilities::indexer::internal::Indexer>
88	indexer_)	127✔
89	: default_buffer_size(DEFAULT_READER_BUFFER_SIZE),	255✔
90	indexer(std::move(indexer_)) {	511✔
91	if (!indexer) {	255✔
92	throw ReaderError(ReaderError::INITIALIZATION_ERROR,	×
93	"Invalid indexer provided");	×
94	}
95	is_open = true;	255✔
96	gz_path = indexer->get_archive_path();	255!
97	index_path = indexer->get_index_path();	256!
98	}	384✔
99
100	GzipReader::~GzipReader() {	1,800✔
101	DFTRACER_UTILS_LOG_DEBUG("Destroying GZIP reader for gz: %s and index: %s",	1,200!
102	gz_path.c_str(), index_path.c_str());
103	reset();	1,200!
104	is_open = false;	1,200✔
105	}	1,800✔
106
107	GzipReader::GzipReader(GzipReader &&other) noexcept	×
108	: gz_path(std::move(other.gz_path)),	×
109	index_path(std::move(other.index_path)),	×
110	is_open(other.is_open),	×
111	default_buffer_size(other.default_buffer_size),	×
112	indexer(std::move(other.indexer)) {	×
113	other.is_open = false;	×
114	}	×
115
116	GzipReader &GzipReader::operator=(GzipReader &&other) noexcept {	×
117	if (this != &other) {	×
118	gz_path = std::move(other.gz_path);	×
119	index_path = std::move(other.index_path);	×
120	is_open = other.is_open;	×
121	default_buffer_size = other.default_buffer_size;	×
122	indexer = std::move(other.indexer);	×
123	other.is_open = false;	×
124	}
125	return *this;	×
126	}
127
128	std::size_t GzipReader::get_max_bytes() const {	373✔
129	check_reader_state(is_open, indexer.get());	373✔
130	std::size_t max_bytes =	187✔
131	static_cast<std::size_t>(indexer.get()->get_max_bytes());	374✔
132	DFTRACER_UTILS_LOG_DEBUG("Maximum bytes available: %zu", max_bytes);	374✔
133	return max_bytes;	374✔
134	}
135
136	std::size_t GzipReader::get_num_lines() const {	110✔
137	check_reader_state(is_open, indexer.get());	110✔
138	std::size_t num_lines = static_cast<std::size_t>(indexer->get_num_lines());	110✔
139	DFTRACER_UTILS_LOG_DEBUG("Total lines available: %zu", num_lines);	110✔
140	return num_lines;	110✔
141	}
142
143	const std::string &GzipReader::get_archive_path() const { return gz_path; }	14✔
144
145	const std::string &GzipReader::get_index_path() const { return index_path; }	8✔
146
147	void GzipReader::set_buffer_size(std::size_t size) {	×
148	default_buffer_size = size;	×
149	}	×
150
151	void GzipReader::reset() {	1,208✔
152	check_reader_state(is_open, indexer.get());	1,208✔
153	stream_cache_.clear();	1,208✔
154	}	1,208✔
155
156	coro::CoroTask<std::size_t> GzipReader::read_async(std::size_t start_bytes,	74,892!
157	std::size_t end_bytes,
158	char *buffer,
159	std::size_t buffer_size) {	12,482!
160	check_reader_state(is_open, indexer.get());	12,482!
161	validate_parameters(buffer, buffer_size, start_bytes, end_bytes,	12,482✔
162	indexer.get()->get_max_bytes());	12,482!
163
164	DFTRACER_UTILS_LOG_DEBUG(	12,470!
165	"GzipReader::read - request: start_bytes=%zu, end_bytes=%zu, "
166	"buffer_size=%zu",
167	start_bytes, end_bytes, buffer_size);
168
169	// Check if we can reuse cached stream
170	if (!stream_cache_.can_continue(StreamType::BYTES, gz_path, start_bytes,	24,940!
171	end_bytes)) {	12,470✔
172	DFTRACER_UTILS_LOG_DEBUG("%s",	161!
173	"GzipReader::read - creating new byte stream");
174	auto new_stream = stream(StreamConfig()	322!
175	.stream_type(StreamType::BYTES)	161!
176	.range_type(RangeType::BYTE_RANGE)	161!
177	.from(start_bytes)	161!
178	.to(end_bytes));	161!
179	stream_cache_.update(std::move(new_stream), StreamType::BYTES, gz_path,	322!
180	start_bytes, end_bytes);	161✔
181	} else {	161✔
182	DFTRACER_UTILS_LOG_DEBUG(	12,309!
183	"%s", "GzipReader::read - reusing cached byte stream");
184	}
185
186	std::size_t result =	24,940✔
187	co_await stream_cache_.get()->read_async(buffer, buffer_size);	37,422!
188	DFTRACER_UTILS_LOG_DEBUG("GzipReader::read - returned %zu bytes", result);	12,470!
189
190	// Update position for next potential read
191	stream_cache_.update_position(start_bytes + result);	12,470!
192
193	co_return result;	12,470!
194	}	62,386!
195
196	coro::CoroTask<std::size_t> GzipReader::read_line_bytes_async(	518,796!
197	std::size_t start_bytes, std::size_t end_bytes, char *buffer,
198	std::size_t buffer_size) {	64,849!
199	check_reader_state(is_open, indexer.get());	194,547✔
200
201	if (end_bytes > indexer.get()->get_max_bytes()) {	64,849!
202	end_bytes = indexer.get()->get_max_bytes();	×
203	}
204
205	validate_parameters(buffer, buffer_size, start_bytes, end_bytes,	64,849!
206	indexer.get()->get_max_bytes());	64,849!
207
208	// Check if we can reuse cached stream
209	if (!stream_cache_.can_continue(StreamType::MULTI_LINES_BYTES, gz_path,	129,698✔
210	start_bytes, end_bytes)) {	64,849✔
211	auto new_stream = stream(StreamConfig()	2,032!
212	.stream_type(StreamType::MULTI_LINES_BYTES)	1,016!
213	.range_type(RangeType::BYTE_RANGE)	1,016!
214	.from(start_bytes)	1,016!
215	.to(end_bytes));	1,016!
216	stream_cache_.update(std::move(new_stream),	2,032!
217	StreamType::MULTI_LINES_BYTES, gz_path,	1,016✔
218	start_bytes, end_bytes);	1,016✔
219	}	1,016✔
220
221	std::size_t result =	259,396✔
222	co_await stream_cache_.get()->read_async(buffer, buffer_size);	324,245!
223
224	// Update position for next potential read
225	stream_cache_.update_position(start_bytes + result);	64,849!
226
227	co_return result;	64,849!
228	}	843,045!
229
230	coro::CoroTask<std::string> GzipReader::read_lines_async(std::size_t start_line,	2,006!
231	std::size_t end_line) {	163!
232	check_reader_state(is_open, indexer.get());	163!
233
234	if (start_line == 0 \|\| end_line == 0) {	163✔
235	throw ReaderError(ReaderError::INVALID_ARGUMENT,	6!
236	"Line numbers must be 1-based (start from 1)");	6!
237	}
238
239	if (start_line > end_line) {	157✔
240	throw ReaderError(ReaderError::INVALID_ARGUMENT,	3!
241	"Start line must be <= end line");	3!
242	}
243
244	std::size_t total_lines = indexer.get()->get_num_lines();	154!
245	if (start_line > total_lines \|\| end_line > total_lines) {	154✔
246	throw ReaderError(ReaderError::INVALID_ARGUMENT,	4!
247	"Line numbers exceed total lines in file (" +	2!
248	std::to_string(total_lines) + ")");	4!
249	}
250
251	// Check if we can reuse cached stream
252	if (!stream_cache_.can_continue(StreamType::MULTI_LINES, gz_path,	304!
253	start_line, end_line)) {	152✔
254	auto new_stream = stream(StreamConfig()	304!
255	.stream_type(StreamType::MULTI_LINES)	152!
256	.range_type(RangeType::LINE_RANGE)	152!
257	.from(start_line)	152!
258	.to(end_line));	152!
259	stream_cache_.update(std::move(new_stream), StreamType::MULTI_LINES,	304!
260	gz_path, start_line, end_line);	152✔
261	}	152✔
262
263	std::string result;	152✔
264	// Pre-allocate to avoid reallocations (like old StringLineProcessor)
265	std::size_t estimated_lines = end_line - start_line + 1;	152✔
266	result.reserve(estimated_lines * 100); // Estimate ~100 bytes per line	152!
267
268	std::vector<char> buffer(default_buffer_size);	152!
269
270	while (!stream_cache_.get()->done()) {	561!
271	std::size_t bytes_read = co_await stream_cache_.get()->read_async(	1,799!
272	buffer.data(), buffer.size());	409✔
273	if (bytes_read == 0) break;	409!
274
275	result.append(buffer.data(), bytes_read);	409!
276	}	409✔
277
278	co_return result;	1,180!
279	}	3,467!
280
281	coro::CoroTask<void> GzipReader::read_lines_with_processor_async(	×
282	std::size_t start_line, std::size_t end_line, LineProcessor &processor) {	×
283	check_reader_state(is_open, indexer.get());	×
284
285	if (start_line == 0 \|\| end_line == 0) {	×
286	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
287	"Line numbers must be 1-based (start from 1)");	×
288	}
289
290	if (start_line > end_line) {	×
291	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
292	"Start line must be <= end line");	×
293	}
294
295	std::size_t total_lines = indexer.get()->get_num_lines();	×
296	if (start_line > total_lines \|\| end_line > total_lines) {	×
297	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
298	"Line numbers exceed total lines in file (" +	×
299	std::to_string(total_lines) + ")");	×
300	}
301
302	processor.begin(start_line, end_line);	×
303
304	// Create a LineStream that returns one line at a time
305	auto line_stream = stream(StreamConfig()	×
306	.stream_type(StreamType::LINE)	×
307	.range_type(RangeType::LINE_RANGE)	×
308	.from(start_line)	×
309	.to(end_line));	×
310
311	std::vector<char> buffer(default_buffer_size);	×
312
313	while (!line_stream->done()) {	×
314	std::size_t bytes_read =
315	co_await line_stream->read_async(buffer.data(), buffer.size());	×
316	if (bytes_read == 0) break;	×
317
318	// LineStream returns one complete line with \n
319	// Processor expects line without \n
320	std::size_t line_length = bytes_read;
321	if (line_length > 0 && buffer[line_length - 1] == '\n') {	×
322	line_length--;
323	}
324
325	if (!co_await processor.process(buffer.data(), line_length)) {	×
326	processor.end();	×
327	co_return;
328	}
329	}	×
330
331	processor.end();	×
332	}	×
333
334	coro::CoroTask<void> GzipReader::read_line_bytes_with_processor_async(	×
335	std::size_t start_bytes, std::size_t end_bytes, LineProcessor &processor) {	×
336	check_reader_state(is_open, indexer.get());	×
337
338	if (end_bytes > indexer.get()->get_max_bytes()) {	×
339	end_bytes = indexer.get()->get_max_bytes();	×
340	}
341
342	if (start_bytes >= end_bytes) {	×
343	co_return;
344	}
345
346	processor.begin(start_bytes, end_bytes);	×
347
348	auto lines_stream = stream(StreamConfig()	×
349	.stream_type(StreamType::LINE_BYTES)	×
350	.range_type(RangeType::BYTE_RANGE)	×
351	.from(start_bytes)	×
352	.to(end_bytes));	×
353
354	std::vector<char> buffer(default_buffer_size);	×
355
356	while (!lines_stream->done()) {	×
357	std::size_t bytes_read =
358	co_await lines_stream->read_async(buffer.data(), buffer.size());	×
359	if (bytes_read == 0) break;	×
360	co_await processor.process(buffer.data(), bytes_read);	×
361	}	×
362
363	processor.end();	×
364	}	×
365
366	bool GzipReader::is_valid() const { return is_open && indexer.get(); }	74!
367
368	std::string GzipReader::get_format_name() const { return "GZIP"; }	2!
369
370	std::unique_ptr<ReaderStream> GzipReader::stream(const StreamConfig &config) {	3,570✔
371	check_reader_state(is_open, indexer.get());	3,570!
372
373	// Extract config parameters
374	StreamType stream_type = config.stream_type();	3,574✔
375	RangeType range_type = config.range_type();	3,575✔
376	std::size_t start = config.start();	3,573✔
377	std::size_t end = config.end();	3,573✔
378	std::size_t buffer_size = config.buffer_size();	3,574✔
379	bool extend_to_line_boundary = config.extend_to_line_boundary();	3,572✔
380
381	// Convert line range to byte range if needed
382	std::size_t start_bytes = start;	3,573✔
383	std::size_t end_bytes = end;	3,573✔
384	std::size_t actual_start_line =	3,573✔
385	1; // Track what line number start_bytes corresponds to
386
387	if (range_type == RangeType::LINE_RANGE) {	3,573✔
388	// Convert line numbers to byte offsets using checkpoints
389	if (start == 0 \|\| end == 0) {	451!
390	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
391	"Line numbers must be 1-based (start from 1)");	×
392	}
393	if (start > end) {	451✔
394	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
395	"Start line must be <= end line");	×
396	}
397
398	std::size_t total_lines = indexer->get_num_lines();	451!
399	if (start > total_lines \|\| end > total_lines) {	452!
400	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
401	"Line numbers exceed total lines in file (" +	×
402	std::to_string(total_lines) + ")");	×
403	}
404
405	// Get checkpoints for the line range
406	std::vector<
407	dftracer::utils::utilities::indexer::internal::IndexerCheckpoint>
408	checkpoints = indexer->get_checkpoints_for_line_range(start, end);	452!
409
410	DFTRACER_UTILS_LOG_DEBUG("Line range %zu-%zu: found %zu checkpoints",	451!
411	start, end, checkpoints.size());
412
413	if (checkpoints.empty()) {	452✔
414	// No checkpoints, read from beginning
415	start_bytes = 0;	180✔
416	end_bytes = indexer->get_max_bytes();	180!
417	actual_start_line = 1;	180✔
418	DFTRACER_UTILS_LOG_DEBUG(	180!
419	"No checkpoints found, using full file: start_bytes=%zu, "
420	"end_bytes=%zu, max_bytes=%" PRIu64,
421	start_bytes, end_bytes, indexer->get_max_bytes());
422	} else {	90✔
423	// Use checkpoint to determine byte range.
424	//
425	// Checkpoint uc_offset values fall at deflate block boundaries
426	// which may land in the middle of a text line. When we start
427	// decompressing from such a mid-line position the first "line"
428	// seen by MultiLineStream is a partial fragment. If
429	// actual_start_line == start_line the fragment is emitted as
430	// the requested first line, producing wrong content.
431	//
432	// To avoid this we choose a checkpoint whose last_line_num is
433	// strictly less than (start - 1), guaranteeing
434	// actual_start_line < start. MultiLineStream then filters
435	// out the (potentially partial) early lines before reaching
436	// the requested range.
437	auto all_checkpoints = indexer->get_checkpoints();	272!
438	bool found_start = false;	272✔
439
440	// Walk checkpoints from the end to find the latest one whose
441	// line range ends before (start - 1).
442	for (auto it = all_checkpoints.rbegin();	14,044✔
443	it != all_checkpoints.rend(); ++it) {	20,227!
444	if (it->last_line_num < start - 1) {	13,277!
445	start_bytes = it->uc_offset;	140!
446	actual_start_line = it->last_line_num + 1;	140!
447	found_start = true;	140✔
448	break;	140✔
449	}
450	}	6,818✔
451
452	if (!found_start) {	272✔
453	// No suitable checkpoint found -- start from beginning
454	start_bytes = 0;	132✔
455	actual_start_line = 1;	132✔
456	}	66✔
457
458	const auto &last_checkpoint = checkpoints.back();	272✔
459	end_bytes = last_checkpoint.uc_offset + last_checkpoint.uc_size;	272✔
460
461	DFTRACER_UTILS_LOG_DEBUG(	272!
462	"Using checkpoints: matched_first_idx=%" PRIu64
463	" "
464	"(first_line=%" PRIu64 ", last_line=%" PRIu64
465	"), "
466	"end_checkpoint_idx=%" PRIu64 " (first_line=%" PRIu64
467	", last_line=%" PRIu64
468	"), "
469	"byte_range=%zu-%zu, actual_start_line=%zu",
470	checkpoints[0].checkpoint_idx, checkpoints[0].first_line_num,
471	checkpoints[0].last_line_num, last_checkpoint.checkpoint_idx,
472	last_checkpoint.first_line_num, last_checkpoint.last_line_num,
473	start_bytes, end_bytes, actual_start_line);
474	}	272✔
475	}	452✔
476
477	// Create appropriate stream type
478	switch (stream_type) {	3,574!
479	case StreamType::BYTES: {	193✔
480	auto byte_stream = std::make_unique<GzipByteStream>(buffer_size);	386!
481	byte_stream->initialize(gz_path, start_bytes, end_bytes, *indexer);	386!
482	return byte_stream;	386✔
483	}	386✔
484	case StreamType::LINE_BYTES: {	159✔
485	// Single line-aligned bytes at a time
486	auto line_byte_stream =
487	std::make_unique<GzipLineByteStream>(buffer_size);	319!
488	line_byte_stream->set_extend_to_line_boundary(	480!
489	extend_to_line_boundary);	160✔
490	line_byte_stream->initialize(gz_path, start_bytes, end_bytes,	639!
491	*indexer);	320✔
492
493	// Wrap with LineStream to return one line-aligned chunk at a time
494	if (range_type == RangeType::LINE_RANGE) {	320✔
495	return std::make_unique<LineStream>(	6!
496	std::move(line_byte_stream), start, end, actual_start_line);	4✔
497	} else {
498	return std::make_unique<LineStream>(	473!
499	std::move(line_byte_stream));	315✔
500	}
501	}	320✔
502	case StreamType::MULTI_LINES_BYTES: {	1,199✔
503	// Multiple line-aligned bytes per read
504	auto line_byte_stream =
505	std::make_unique<GzipLineByteStream>(buffer_size);	2,399!
506	line_byte_stream->set_extend_to_line_boundary(	3,599!
507	extend_to_line_boundary);	1,200✔
508	line_byte_stream->initialize(gz_path, start_bytes, end_bytes,	4,800!
509	*indexer);	2,400✔
510	return line_byte_stream;	2,399✔
511	}	2,398✔
512	case StreamType::LINE: {	54✔
513	// Single parsed line per read
514	auto line_byte_stream =
515	std::make_unique<GzipLineByteStream>(buffer_size);	108!
516	line_byte_stream->set_extend_to_line_boundary(	162!
517	extend_to_line_boundary);	54✔
518	line_byte_stream->initialize(gz_path, start_bytes, end_bytes,	216!
519	*indexer);	108✔
520
521	if (range_type == RangeType::LINE_RANGE) {	108✔
522	return std::make_unique<LineStream>(	159!
523	std::move(line_byte_stream), start, end, actual_start_line);	106✔
524	} else {
525	return std::make_unique<LineStream>(	3!
526	std::move(line_byte_stream));	2✔
527	}
528	}	108✔
529	case StreamType::MULTI_LINES: {	181✔
530	// Multiple parsed lines per read
531	auto line_byte_stream =
532	std::make_unique<GzipLineByteStream>(buffer_size);	362!
533	line_byte_stream->set_extend_to_line_boundary(	543!
534	extend_to_line_boundary);	181✔
535	line_byte_stream->initialize(gz_path, start_bytes, end_bytes,	724!
536	*indexer);	362✔
537
538	if (range_type == RangeType::LINE_RANGE) {	362✔
539	return std::make_unique<MultiLineStream>(	507!
540	std::move(line_byte_stream), start, end, actual_start_line);	338✔
541	} else {
542	return std::make_unique<MultiLineStream>(	36!
543	std::move(line_byte_stream));	24✔
544	}
545	}	362✔
546	default:
547	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
548	"Invalid stream type");	×
549	}
550	}	1,788✔
551
552	} // namespace dftracer::utils::utilities::reader::internal

llnl / dftracer-utils / 29043022308

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