reader.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright 2026 Johnson Ogundeji
#pragma once
 
// ---------------------------------------------------------------------------
// ParquetReader — Parquet file reader
//
// Opens a Parquet file, verifies PAR1 magic bytes, deserializes the Thrift
// footer to extract FileMetaData, builds a Schema, and provides typed access
// to column data via ColumnReader.
//
// Supports:
//   - Typed column reads (read_column<T>)
//   - String conversion reads (read_column_as_strings)
//   - Full row-group and file reads
//   - Column projection by name
//   - Per-column statistics access
//   - Multiple encodings: PLAIN, RLE_DICTIONARY, DELTA_BINARY_PACKED,
//     BYTE_STREAM_SPLIT, RLE (booleans)
//   - Decompression via CodecRegistry (Snappy, ZSTD, LZ4, etc.)
// ---------------------------------------------------------------------------
 
#include "signet/types.hpp"
#include "signet/error.hpp"
#include "signet/schema.hpp"
#include "signet/column_reader.hpp"
#include "signet/memory.hpp"
#include "signet/thrift/compact.hpp"
#include "signet/thrift/types.hpp"
#include "signet/encoding/rle.hpp"
#include "signet/encoding/dictionary.hpp"
#include "signet/encoding/delta.hpp"
#include "signet/encoding/byte_stream_split.hpp"
#include "signet/compression/codec.hpp"
#include "signet/compression/snappy.hpp"
#include "signet/compression/zstd.hpp"
#include "signet/compression/lz4.hpp"
#include "signet/compression/gzip.hpp"
#include "signet/bloom/split_block.hpp"
#include "signet/column_index.hpp"
 
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
#include "signet/crypto/pme.hpp"
#endif
 
#include <array>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <filesystem>
#include <fstream>
#include <functional>
#include <mutex>
#include <stdexcept>
#include <memory>
#include <optional>
#include <string>
#include <vector>
 
// Windows macro pollution: undefine after all includes.
// TIME_MS from <mmsystem.h>, OPTIONAL from <sal.h>
#ifdef TIME_MS
#undef TIME_MS
#endif
#ifdef OPTIONAL
#undef OPTIONAL
#endif
 
namespace signet::forge {
 
namespace detail_reader {
 
inline uint32_t crc32(const void* data, size_t length) noexcept {
    static constexpr auto make_table = []() {
        std::array<uint32_t, 256> t{};
        for (uint32_t i = 0; i < 256; ++i) {
            uint32_t c = i;
            for (int k = 0; k < 8; ++k)
                c = (c & 1u) ? (0xEDB88320u ^ (c >> 1)) : (c >> 1);
            t[i] = c;
        }
        return t;
    };
    static constexpr auto table = make_table();
    uint32_t crc = 0xFFFFFFFFu;
    auto* p = static_cast<const uint8_t*>(data);
    for (size_t i = 0; i < length; ++i)
        crc = table[(crc ^ p[i]) & 0xFFu] ^ (crc >> 8);
    return crc ^ 0xFFFFFFFFu;
}
 
} // namespace detail_reader
 
static constexpr size_t PARQUET_MAX_PAGE_SIZE = 256ULL * 1024ULL * 1024ULL;  // 256 MB hard cap per page
 
static constexpr int64_t MAX_VALUES_PER_PAGE = 100'000'000;  // 100M values per page — OOM guard
 
[[nodiscard]] inline expected<size_t> validate_page_value_count(
        int64_t num_values,
        const char* context) {
    if (num_values < 0 || num_values > MAX_VALUES_PER_PAGE) {
        return Error{ErrorCode::CORRUPT_PAGE,
                     std::string(context) + ": num_values out of valid range"};
    }
    return static_cast<size_t>(num_values);
}
 
inline constexpr uint8_t kEncryptedPageHeaderMagic[4] = {'S', 'P', 'H', '1'};
 
[[nodiscard]] inline bool has_encrypted_page_header_prefix(
        const uint8_t* data,
        size_t size) noexcept {
    return size >= 8 &&
        std::memcmp(data, kEncryptedPageHeaderMagic, sizeof(kEncryptedPageHeaderMagic)) == 0;
}
 
[[nodiscard]] inline uint32_t load_le32(const uint8_t* data) noexcept {
    return static_cast<uint32_t>(data[0]) |
           (static_cast<uint32_t>(data[1]) << 8) |
           (static_cast<uint32_t>(data[2]) << 16) |
           (static_cast<uint32_t>(data[3]) << 24);
}
 
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
[[nodiscard]] inline bool decryptor_has_column_key(
        const crypto::EncryptionConfig& config,
        const std::string& col_name) {
    if (!config.default_column_key.empty()) {
        return true;
    }
    return std::any_of(config.column_keys.begin(), config.column_keys.end(),
        [&](const auto& ck) {
            return ck.column_name == col_name && !ck.key.empty();
        });
}
#endif
 
class ParquetReader {
public:
    // ===================================================================
    // Open a Parquet file
    // ===================================================================
 
    static expected<ParquetReader> open(
        const std::filesystem::path& path
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
        , const std::optional<crypto::EncryptionConfig>& encryption = std::nullopt
#endif
        ) {
 
        // Ensure built-in codecs are available for page decompression.
        ensure_default_codecs_registered();
 
        // --- Read entire file into memory ---
        std::error_code ec;
        auto file_size = std::filesystem::file_size(path, ec);
        if (ec) {
            return Error{ErrorCode::IO_ERROR,
                         "cannot determine file size: " + path.string() +
                         " (" + ec.message() + ")"};
        }
 
        static constexpr int64_t MAX_FILE_SIZE = INT64_C(4) * 1024 * 1024 * 1024; // 4 GB
        if (static_cast<int64_t>(file_size) > MAX_FILE_SIZE) {
            return Error{ErrorCode::INVALID_ARGUMENT,
                         "File exceeds 4 GB limit; use MmapParquetReader for large files"};
        }
 
        if (file_size < 12) {
            // Minimum: 4 (magic) + 4 (footer len) + 4 (magic) = 12
            return Error{ErrorCode::INVALID_FILE,
                         "file too small to be a valid Parquet file: " +
                         path.string()};
        }
 
        std::ifstream ifs(path, std::ios::binary);
        if (!ifs) {
            return Error{ErrorCode::IO_ERROR,
                         "cannot open file: " + path.string()};
        }
 
        std::vector<uint8_t> file_data(static_cast<size_t>(file_size));
        ifs.read(reinterpret_cast<char*>(file_data.data()),
                 static_cast<std::streamsize>(file_size));
        if (!ifs) {
            return Error{ErrorCode::IO_ERROR,
                         "failed to read file: " + path.string()};
        }
        ifs.close();
 
        // --- Verify PAR1 magic at start ---
        const size_t sz = file_data.size();
 
        uint32_t magic_start;
        std::memcpy(&magic_start, file_data.data(), 4);
        if (magic_start != PARQUET_MAGIC) {
            return Error{ErrorCode::INVALID_FILE,
                         "missing PAR1 magic at start of file"};
        }
 
        // --- Check trailing magic: PAR1 (plaintext) or PARE (encrypted footer) ---
        uint32_t magic_end;
        std::memcpy(&magic_end, file_data.data() + sz - 4, 4);
 
        bool encrypted_footer = (magic_end == PARQUET_MAGIC_ENCRYPTED);
 
        if (magic_end != PARQUET_MAGIC && magic_end != PARQUET_MAGIC_ENCRYPTED) {
            return Error{ErrorCode::INVALID_FILE,
                         "missing PAR1/PARE magic at end of file"};
        }
 
        // --- Read footer length (4-byte LE uint32 at [size-8, size-4]) ---
        uint32_t footer_len;
        std::memcpy(&footer_len, file_data.data() + sz - 8, 4);
 
        if (footer_len == 0 || static_cast<size_t>(footer_len) > sz - 12) {
            return Error{ErrorCode::CORRUPT_FOOTER,
                         "invalid footer length: " + std::to_string(footer_len)};
        }
 
        // --- Prepare decryptor if needed ---
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
        std::unique_ptr<crypto::FileDecryptor> decryptor;
        if (encrypted_footer || encryption) {
            if (!encryption) {
                return Error{ErrorCode::ENCRYPTION_ERROR,
                             "file has encrypted footer (PARE magic) but no "
                             "encryption config was provided"};
            }
            decryptor = std::make_unique<crypto::FileDecryptor>(*encryption);
        }
#else
        if (encrypted_footer) {
            return Error{ErrorCode::LICENSE_ERROR,
                         "encrypted footer (PARE) requires commercial build and license"};
        }
#endif
 
        // --- Deserialize FileMetaData from footer ---
        size_t footer_offset = sz - 8 - footer_len;
        const uint8_t* footer_ptr = file_data.data() + footer_offset;
        size_t footer_size = footer_len;
 
        // If footer is encrypted, decrypt it first
        std::vector<uint8_t> decrypted_footer_buf;
        if (encrypted_footer) {
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
            auto dec_result = decryptor->decrypt_footer(footer_ptr, footer_size);
            if (!dec_result) return dec_result.error();
            decrypted_footer_buf = std::move(*dec_result);
            footer_ptr  = decrypted_footer_buf.data();
            footer_size = decrypted_footer_buf.size();
#else
            return Error{ErrorCode::LICENSE_ERROR,
                         "encrypted footer (PARE) requires commercial build and license"};
#endif
        }
 
        thrift::CompactDecoder dec(footer_ptr, footer_size);
 
        thrift::FileMetaData metadata;
        if (auto r = metadata.deserialize(dec); !r.has_value()) {
            return r.error();
        }
 
        // --- Build Schema from FileMetaData.schema ---
        // The first element is the root (group) node — skip it.
        // Subsequent leaf elements become columns.
        std::string schema_name;
        std::vector<ColumnDescriptor> columns;
 
        if (!metadata.schema.empty()) {
            schema_name = metadata.schema[0].name;
 
            for (size_t i = 1; i < metadata.schema.size(); ++i) {
                const auto& elem = metadata.schema[i];
 
                // Skip group nodes (those with num_children set)
                if (elem.num_children.has_value()) {
                    continue;
                }
 
                ColumnDescriptor cd;
                cd.name = elem.name;
                cd.physical_type = elem.type.value_or(PhysicalType::BYTE_ARRAY);
                cd.repetition = elem.repetition_type.value_or(Repetition::REQUIRED);
 
                if (elem.type_length.has_value()) {
                    cd.type_length = *elem.type_length;
                }
                if (elem.precision.has_value()) {
                    cd.precision = *elem.precision;
                }
                if (elem.scale.has_value()) {
                    cd.scale = *elem.scale;
                }
 
                // Map ConvertedType to LogicalType for common cases
                if (elem.converted_type.has_value()) {
                    cd.logical_type = converted_type_to_logical(*elem.converted_type);
                }
 
                columns.push_back(std::move(cd));
            }
        }
 
        // --- Assemble the reader ---
        ParquetReader reader;
        reader.file_data_  = std::move(file_data);
        reader.metadata_   = std::move(metadata);
        reader.schema_     = Schema(std::move(schema_name), std::move(columns));
        reader.created_by_ = reader.metadata_.created_by.value_or("");
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
        reader.decryptor_  = std::move(decryptor);
#endif
 
        return reader;
    }
 
    // ===================================================================
    // File metadata accessors
    // ===================================================================
 
    [[nodiscard]] const Schema& schema() const { return schema_; }
 
    [[nodiscard]] int64_t num_rows() const { return metadata_.num_rows; }
 
    [[nodiscard]] int64_t num_row_groups() const {
        return static_cast<int64_t>(metadata_.row_groups.size());
    }
 
    [[nodiscard]] const std::string& created_by() const { return created_by_; }
 
    [[nodiscard]] const std::vector<thrift::KeyValue>& key_value_metadata() const {
        static const std::vector<thrift::KeyValue> empty;
        return metadata_.key_value_metadata.has_value()
                   ? *metadata_.key_value_metadata
                   : empty;
    }
 
    // ===================================================================
    // Row group info
    // ===================================================================
 
    struct RowGroupInfo {
        int64_t num_rows;         
        int64_t total_byte_size;  
        int64_t row_group_index;  
    };
 
    RowGroupInfo row_group(size_t index) const {
        if (index >= metadata_.row_groups.size()) {
            throw std::out_of_range("ParquetReader::row_group: index " +
                std::to_string(index) + " >= " +
                std::to_string(metadata_.row_groups.size()));
        }
        const auto& rg = metadata_.row_groups[index];
        return {rg.num_rows,
                rg.total_byte_size,
                static_cast<int64_t>(index)};
    }
 
    // ===================================================================
    // File statistics (aggregate metadata for the entire file)
    // ===================================================================
 
    [[nodiscard]] FileStats file_stats() const {
        FileStats fs;
        fs.file_size_bytes = static_cast<int64_t>(file_data_.size());
        fs.total_rows      = metadata_.num_rows;
        fs.num_row_groups  = static_cast<int64_t>(metadata_.row_groups.size());
        fs.num_columns     = static_cast<int64_t>(schema_.num_columns());
        fs.created_by      = created_by_;
 
        // Aggregate per-column stats across all row groups
        fs.columns.resize(schema_.num_columns());
        for (size_t c = 0; c < schema_.num_columns(); ++c) {
            auto& col = fs.columns[c];
            col.column_name  = schema_.column(c).name;
            col.physical_type = schema_.column(c).physical_type;
            col.logical_type  = schema_.column(c).logical_type;
        }
 
        int64_t total_uncompressed = 0;
        int64_t total_compressed   = 0;
 
        for (const auto& rg : metadata_.row_groups) {
            for (size_t c = 0; c < rg.columns.size() && c < schema_.num_columns(); ++c) {
                const auto& cc = rg.columns[c];
                if (!cc.meta_data.has_value()) continue;
                const auto& cmd = *cc.meta_data;
                auto& col = fs.columns[c];
 
                col.uncompressed_bytes += cmd.total_uncompressed_size;
                col.compressed_bytes   += cmd.total_compressed_size;
                col.num_values         += cmd.num_values;
                col.compression         = cmd.codec;
 
                if (cmd.statistics.has_value() && cmd.statistics->null_count.has_value()) {
                    col.null_count += *cmd.statistics->null_count;
                }
 
                // Check bloom filter presence
                if (cc.bloom_filter_offset.has_value() && *cc.bloom_filter_offset >= 0) {
                    col.has_bloom_filter = true;
                }
                // Check page index presence
                if (cc.column_index_offset.has_value() && *cc.column_index_offset >= 0) {
                    col.has_page_index = true;
                }
 
                total_uncompressed += cmd.total_uncompressed_size;
                total_compressed   += cmd.total_compressed_size;
            }
        }
 
        if (total_compressed > 0) {
            fs.compression_ratio = static_cast<double>(total_uncompressed)
                                 / static_cast<double>(total_compressed);
        }
        if (fs.total_rows > 0) {
            fs.bytes_per_row = static_cast<double>(fs.file_size_bytes)
                             / static_cast<double>(fs.total_rows);
        }
 
        return fs;
    }
 
    static void ensure_default_codecs_registered() {
        static std::once_flag codec_flag;
        std::call_once(codec_flag, [] {
            register_snappy_codec();
#ifdef SIGNET_HAS_ZSTD
            register_zstd_codec();
#endif
#ifdef SIGNET_HAS_LZ4
            register_lz4_codec();
#endif
#ifdef SIGNET_HAS_GZIP
            register_gzip_codec();
#endif
        });
    }
 
    // ===================================================================
    // Read a single column from a row group as a typed vector
    // ===================================================================
 
    template <typename T>
    expected<std::vector<T>> read_column(size_t row_group_index,
                                          size_t column_index) {
        // --- Validate indices ---
        if (row_group_index >= metadata_.row_groups.size()) {
            return Error{ErrorCode::OUT_OF_RANGE,
                         "row group index out of range"};
        }
        if (column_index >= schema_.num_columns()) {
            return Error{ErrorCode::OUT_OF_RANGE,
                         "column index out of range"};
        }
 
        const auto& rg = metadata_.row_groups[row_group_index];
        if (column_index >= rg.columns.size()) {
            return Error{ErrorCode::OUT_OF_RANGE,
                         "column index out of range"};
        }
 
        const auto& chunk = rg.columns[column_index];
        if (!chunk.meta_data.has_value()) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "column chunk has no metadata"};
        }
        const auto& col_meta = *chunk.meta_data;
 
        // --- Detect encoding strategy from column metadata ---
        bool has_dict = false;
        Encoding data_encoding = Encoding::PLAIN;
 
        for (auto enc : col_meta.encodings) {
            if (enc == Encoding::PLAIN_DICTIONARY || enc == Encoding::RLE_DICTIONARY) {
                has_dict = true;
            }
            if (enc == Encoding::DELTA_BINARY_PACKED) {
                data_encoding = Encoding::DELTA_BINARY_PACKED;
            }
            if (enc == Encoding::BYTE_STREAM_SPLIT) {
                data_encoding = Encoding::BYTE_STREAM_SPLIT;
            }
            if (enc == Encoding::RLE && col_meta.type == PhysicalType::BOOLEAN) {
                data_encoding = Encoding::RLE;
            }
        }
 
        // Resolve column name for encryption context
        const std::string& col_name = col_meta.path_in_schema.empty()
            ? schema_.column(column_index).name
            : col_meta.path_in_schema[0];
 
        // --- Dictionary encoding path ---
        // Dictionary encoding supports: string, int32, int64, float, double
        if (has_dict) {
            if constexpr (std::is_same_v<T, std::string> ||
                          std::is_same_v<T, int32_t> ||
                          std::is_same_v<T, int64_t> ||
                          std::is_same_v<T, float> ||
                          std::is_same_v<T, double>) {
                return read_column_dict<T>(col_meta, col_name,
                    static_cast<int32_t>(row_group_index));
            } else {
                return Error{ErrorCode::UNSUPPORTED_ENCODING,
                             "dictionary encoding not supported for this type"};
            }
        }
 
        // --- DELTA_BINARY_PACKED path (INT32/INT64 only) ---
        if (data_encoding == Encoding::DELTA_BINARY_PACKED) {
            if constexpr (std::is_same_v<T, int32_t> ||
                          std::is_same_v<T, int64_t>) {
                return read_column_delta<T>(col_meta, col_name,
                    static_cast<int32_t>(row_group_index));
            } else {
                return Error{ErrorCode::UNSUPPORTED_ENCODING,
                             "DELTA_BINARY_PACKED only supports INT32/INT64"};
            }
        }
 
        // --- BYTE_STREAM_SPLIT path (FLOAT/DOUBLE only) ---
        if (data_encoding == Encoding::BYTE_STREAM_SPLIT) {
            if constexpr (std::is_same_v<T, float> ||
                          std::is_same_v<T, double>) {
                return read_column_bss<T>(col_meta, col_name,
                    static_cast<int32_t>(row_group_index));
            } else {
                return Error{ErrorCode::UNSUPPORTED_ENCODING,
                             "BYTE_STREAM_SPLIT only supports FLOAT/DOUBLE"};
            }
        }
 
        // --- RLE path (boolean only) ---
        if (data_encoding == Encoding::RLE &&
            col_meta.type == PhysicalType::BOOLEAN) {
            if constexpr (std::is_same_v<T, bool>) {
                return read_column_rle_bool<T>(col_meta, col_name,
                    static_cast<int32_t>(row_group_index));
            } else {
                return Error{ErrorCode::UNSUPPORTED_ENCODING,
                             "RLE boolean encoding requires bool type"};
            }
        }
 
        // --- Default: PLAIN encoding via ColumnReader ---
        auto reader_result = make_column_reader(row_group_index, column_index);
        if (!reader_result) return reader_result.error();
 
        auto& [col_reader, num_values] = reader_result.value();
        auto count_result = validate_page_value_count(num_values, "ParquetReader RLE_BOOL");
        if (!count_result) return count_result.error();
        size_t count = *count_result;
 
        if constexpr (std::is_same_v<T, bool>) {
            // std::vector<bool> has no .data() — read one by one
            std::vector<bool> values;
            values.reserve(count);
            for (size_t i = 0; i < count; ++i) {
                auto val = col_reader.template read<bool>();
                if (!val) return val.error();
                values.push_back(*val);
            }
            return values;
        } else {
            std::vector<T> values(count);
            auto batch_result = col_reader.template read_batch<T>(
                values.data(), count);
            if (!batch_result) return batch_result.error();
            return values;
        }
    }
 
    // ===================================================================
    // Read column as strings (converts any type to string representation)
    // ===================================================================
 
    expected<std::vector<std::string>> read_column_as_strings(
            size_t row_group_index, size_t column_index) {
        if (row_group_index >= metadata_.row_groups.size()) {
            return Error{ErrorCode::OUT_OF_RANGE, "row group index out of range"};
        }
        if (column_index >= schema_.num_columns()) {
            return Error{ErrorCode::OUT_OF_RANGE, "column index out of range"};
        }
 
        PhysicalType pt = schema_.column(column_index).physical_type;
 
        switch (pt) {
        case PhysicalType::BOOLEAN: {
            auto res = read_column<bool>(row_group_index, column_index);
            if (!res) return res.error();
            return to_string_vec(res.value());
        }
        case PhysicalType::INT32: {
            auto res = read_column<int32_t>(row_group_index, column_index);
            if (!res) return res.error();
            return to_string_vec(res.value());
        }
        case PhysicalType::INT64: {
            auto res = read_column<int64_t>(row_group_index, column_index);
            if (!res) return res.error();
            return to_string_vec(res.value());
        }
        case PhysicalType::FLOAT: {
            auto res = read_column<float>(row_group_index, column_index);
            if (!res) return res.error();
            return to_string_vec(res.value());
        }
        case PhysicalType::DOUBLE: {
            auto res = read_column<double>(row_group_index, column_index);
            if (!res) return res.error();
            return to_string_vec(res.value());
        }
        case PhysicalType::BYTE_ARRAY: {
            // Strings are already strings — read directly
            return read_column<std::string>(row_group_index, column_index);
        }
        case PhysicalType::FIXED_LEN_BYTE_ARRAY: {
            // Read as raw bytes, hex-encode each value
            auto reader_result = make_column_reader(row_group_index, column_index);
            if (!reader_result) return reader_result.error();
            auto& [col_reader, num_values] = reader_result.value();
 
            std::vector<std::string> result;
            result.reserve(static_cast<size_t>(num_values));
            for (int64_t i = 0; i < num_values; ++i) {
                auto bytes_result = col_reader.read_bytes();
                if (!bytes_result) return bytes_result.error();
                result.push_back(hex_encode(bytes_result.value()));
            }
            return result;
        }
        default:
            return Error{ErrorCode::UNSUPPORTED_TYPE,
                         "unsupported physical type for string conversion"};
        }
    }
 
    // ===================================================================
    // Read all columns from a row group as vectors of strings
    // ===================================================================
 
    expected<std::vector<std::vector<std::string>>> read_row_group(
            size_t row_group_index) {
        size_t num_cols = schema_.num_columns();
        std::vector<std::vector<std::string>> columns(num_cols);
 
        for (size_t c = 0; c < num_cols; ++c) {
            auto res = read_column_as_strings(row_group_index, c);
            if (!res) return res.error();
            columns[c] = std::move(res.value());
        }
 
        return columns;
    }
 
    // ===================================================================
    // Read entire file as vector of rows (each row = vector of strings)
    // ===================================================================
 
    expected<std::vector<std::vector<std::string>>> read_all() {
        size_t num_cols = schema_.num_columns();
        int64_t safe_rows = (metadata_.num_rows < 0) ? 0 : metadata_.num_rows;
        if (static_cast<uint64_t>(safe_rows) > 1024ULL * 1024 * 1024)
            return Error{ErrorCode::INVALID_ARGUMENT, "num_rows exceeds 1 billion limit"};
        std::vector<std::vector<std::string>> rows;
        rows.reserve(static_cast<size_t>(safe_rows));
 
        for (size_t rg = 0; rg < metadata_.row_groups.size(); ++rg) {
            auto cols_result = read_row_group(rg);
            if (!cols_result) return cols_result.error();
 
            const auto& col_data = cols_result.value();
            if (col_data.empty()) continue;
 
            size_t rg_rows = col_data[0].size();
            for (size_t r = 0; r < rg_rows; ++r) {
                std::vector<std::string> row(num_cols);
                for (size_t c = 0; c < num_cols; ++c) {
                    if (r < col_data[c].size()) {
                        row[c] = col_data[c][r];
                    }
                }
                rows.push_back(std::move(row));
            }
        }
 
        return rows;
    }
 
    // ===================================================================
    // Column projection -- read only specific columns by name
    // ===================================================================
 
    expected<std::vector<std::vector<std::string>>> read_columns(
            const std::vector<std::string>& column_names) {
        // Resolve column indices
        std::vector<size_t> indices;
        indices.reserve(column_names.size());
 
        for (const auto& name : column_names) {
            auto idx = schema_.find_column(name);
            if (!idx.has_value()) {
                return Error{ErrorCode::SCHEMA_MISMATCH,
                             "column not found: " + name};
            }
            indices.push_back(*idx);
        }
 
        // Read across all row groups
        size_t proj_cols = indices.size();
        std::vector<std::vector<std::string>> result(proj_cols);
 
        for (size_t rg = 0; rg < metadata_.row_groups.size(); ++rg) {
            for (size_t p = 0; p < proj_cols; ++p) {
                auto col_result = read_column_as_strings(rg, indices[p]);
                if (!col_result) return col_result.error();
 
                auto& col_vec = col_result.value();
                result[p].insert(result[p].end(),
                                 std::make_move_iterator(col_vec.begin()),
                                 std::make_move_iterator(col_vec.end()));
            }
        }
 
        return result;
    }
 
    // ===================================================================
    // Statistics for a column in a row group
    // ===================================================================
 
    const thrift::Statistics* column_statistics(size_t row_group_index,
                                                 size_t column_index) const {
        if (row_group_index >= metadata_.row_groups.size()) return nullptr;
        const auto& rg = metadata_.row_groups[row_group_index];
        if (column_index >= rg.columns.size()) return nullptr;
        const auto& chunk = rg.columns[column_index];
        if (!chunk.meta_data.has_value()) return nullptr;
        if (!chunk.meta_data->statistics.has_value()) return nullptr;
        return &(*chunk.meta_data->statistics);
    }
 
    // ===================================================================
    // Bloom filter access
    // ===================================================================
 
    [[nodiscard]] expected<SplitBlockBloomFilter> read_bloom_filter(
            size_t row_group_index, size_t column_index) const {
        if (row_group_index >= metadata_.row_groups.size()) {
            return Error{ErrorCode::OUT_OF_RANGE,
                         "row group index out of range"};
        }
 
        const auto& rg = metadata_.row_groups[row_group_index];
        if (column_index >= rg.columns.size()) {
            return Error{ErrorCode::OUT_OF_RANGE,
                         "column index out of range"};
        }
 
        const auto& chunk = rg.columns[column_index];
        if (!chunk.bloom_filter_offset.has_value()) {
            return Error{ErrorCode::INVALID_FILE,
                         "no bloom filter for this column chunk"};
        }
 
        int64_t bf_offset = *chunk.bloom_filter_offset;
        if (bf_offset < 0 || static_cast<size_t>(bf_offset) + 4 > file_data_.size()) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "bloom filter offset out of file bounds"};
        }
 
        // Read 4-byte LE size header
        uint32_t bf_size = 0;
        std::memcpy(&bf_size, file_data_.data() + bf_offset, 4);
 
        size_t data_start = static_cast<size_t>(bf_offset) + 4;
        if (data_start + bf_size > file_data_.size()) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "bloom filter data extends past end of file"};
        }
 
        if (bf_size == 0 || (bf_size % SplitBlockBloomFilter::kBytesPerBlock) != 0) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "invalid bloom filter size: " + std::to_string(bf_size)};
        }
 
        return SplitBlockBloomFilter::from_data(
            file_data_.data() + data_start, bf_size);
    }
 
    template <typename T>
    [[nodiscard]] bool bloom_might_contain(
            size_t row_group_index, size_t column_index,
            const T& value) const {
        auto bf_result = read_bloom_filter(row_group_index, column_index);
        if (!bf_result) {
            return true;  // No bloom filter — cannot rule out the value
        }
        return bf_result->might_contain_value(value);
    }
 
    // ===================================================================
    // Page Index access (ColumnIndex + OffsetIndex)
    // ===================================================================
 
    [[nodiscard]] expected<ColumnIndex> read_column_index(
            size_t row_group_index, size_t column_index) const {
        if (row_group_index >= metadata_.row_groups.size())
            return Error{ErrorCode::OUT_OF_RANGE, "row group index out of range"};
        const auto& rg = metadata_.row_groups[row_group_index];
        if (column_index >= rg.columns.size())
            return Error{ErrorCode::OUT_OF_RANGE, "column index out of range"};
 
        const auto& chunk = rg.columns[column_index];
        if (!chunk.column_index_offset.has_value() || !chunk.column_index_length.has_value())
            return Error{ErrorCode::INVALID_FILE, "no column index for this column chunk"};
 
        int64_t ci_offset = *chunk.column_index_offset;
        int32_t ci_length = *chunk.column_index_length;
        if (ci_offset < 0 || ci_length < 0)
            return Error{ErrorCode::CORRUPT_PAGE, "column index offset/length negative"};
        auto uoff = static_cast<size_t>(ci_offset);
        auto ulen = static_cast<size_t>(ci_length);
        if (uoff > file_data_.size() || ulen > file_data_.size() - uoff)
            return Error{ErrorCode::CORRUPT_PAGE, "column index offset/length out of bounds"};
 
        thrift::CompactDecoder dec(file_data_.data() + ci_offset,
                                    static_cast<size_t>(ci_length));
        ColumnIndex ci;
        ci.deserialize(dec);
        if (!dec.good())
            return Error{ErrorCode::CORRUPT_PAGE, "column index deserialization failed"};
        return ci;
    }
 
    [[nodiscard]] expected<OffsetIndex> read_offset_index(
            size_t row_group_index, size_t column_index) const {
        if (row_group_index >= metadata_.row_groups.size())
            return Error{ErrorCode::OUT_OF_RANGE, "row group index out of range"};
        const auto& rg = metadata_.row_groups[row_group_index];
        if (column_index >= rg.columns.size())
            return Error{ErrorCode::OUT_OF_RANGE, "column index out of range"};
 
        const auto& chunk = rg.columns[column_index];
        if (!chunk.offset_index_offset.has_value() || !chunk.offset_index_length.has_value())
            return Error{ErrorCode::INVALID_FILE, "no offset index for this column chunk"};
 
        int64_t oi_offset = *chunk.offset_index_offset;
        int32_t oi_length = *chunk.offset_index_length;
        if (oi_offset < 0 || oi_length < 0)
            return Error{ErrorCode::CORRUPT_PAGE, "offset index offset/length negative"};
        auto uoff2 = static_cast<size_t>(oi_offset);
        auto ulen2 = static_cast<size_t>(oi_length);
        if (uoff2 > file_data_.size() || ulen2 > file_data_.size() - uoff2)
            return Error{ErrorCode::CORRUPT_PAGE, "offset index offset/length out of bounds"};
 
        thrift::CompactDecoder dec(file_data_.data() + oi_offset,
                                    static_cast<size_t>(oi_length));
        OffsetIndex oi;
        oi.deserialize(dec);
        if (!dec.good())
            return Error{ErrorCode::CORRUPT_PAGE, "offset index deserialization failed"};
        return oi;
    }
 
    [[nodiscard]] bool has_page_index(size_t row_group_index, size_t column_index) const {
        if (row_group_index >= metadata_.row_groups.size()) return false;
        const auto& rg = metadata_.row_groups[row_group_index];
        if (column_index >= rg.columns.size()) return false;
        const auto& chunk = rg.columns[column_index];
        return chunk.column_index_offset.has_value() && chunk.offset_index_offset.has_value();
    }
 
    // ===================================================================
    // Special members
    // ===================================================================
 
    ~ParquetReader() = default;
    ParquetReader(ParquetReader&&) noexcept = default;
    ParquetReader& operator=(ParquetReader&&) noexcept = default;
 
private:
    ParquetReader() = default;
 
    std::vector<uint8_t>      file_data_;   
    thrift::FileMetaData      metadata_;    
    Schema                    schema_;      
    std::string               created_by_;  
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
    std::unique_ptr<crypto::FileDecryptor> decryptor_;  
#endif
 
    std::vector<std::vector<uint8_t>> decompressed_buffers_;
 
    struct ColumnReaderWithCount {
        ColumnReader reader;      
        int64_t      num_values;  
    };
 
    expected<ColumnReaderWithCount> make_column_reader(
            size_t row_group_index, size_t column_index) {
        if (row_group_index >= metadata_.row_groups.size()) {
            return Error{ErrorCode::OUT_OF_RANGE,
                         "row group index out of range"};
        }
 
        const auto& rg = metadata_.row_groups[row_group_index];
        if (column_index >= rg.columns.size()) {
            return Error{ErrorCode::OUT_OF_RANGE,
                         "column index out of range"};
        }
 
        const auto& chunk = rg.columns[column_index];
        if (!chunk.meta_data.has_value()) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "column chunk has no metadata"};
        }
 
        const auto& col_meta = *chunk.meta_data;
 
        // Locate the data page in the file buffer.
        // For dictionary-encoded columns, the dictionary page comes first
        // at dictionary_page_offset, and data_page_offset points to the
        // data page (which contains RLE-encoded indices).
        int64_t offset = col_meta.data_page_offset;
        if (offset < 0 || static_cast<size_t>(offset) >= file_data_.size()) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "data_page_offset out of file bounds"};
        }
 
        size_t remaining = file_data_.size() - static_cast<size_t>(offset);
        const uint8_t* page_start = file_data_.data() + offset;
 
        const std::string col_name = col_meta.path_in_schema.empty()
            ? schema_.column(column_index).name
            : col_meta.path_in_schema[0];
 
        thrift::PageHeader page_header;
        size_t header_size = 0;
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
        const bool column_encrypted = decryptor_ &&
            decryptor_has_column_key(decryptor_->config(), col_name);
        if (has_encrypted_page_header_prefix(page_start, remaining)) {
            if (!column_encrypted) {
                return Error{ErrorCode::ENCRYPTION_ERROR,
                             "encrypted page header encountered without matching decryptor for column '" +
                                 col_name + "'"};
            }
            const uint32_t encrypted_header_size = load_le32(page_start + 4);
            if (encrypted_header_size == 0 ||
                static_cast<size_t>(encrypted_header_size) > remaining - 8) {
                return Error{ErrorCode::CORRUPT_PAGE,
                             "encrypted page header size out of range"};
            }
            auto header_result = decryptor_->decrypt_data_page_header(
                page_start + 8, encrypted_header_size, col_name,
                static_cast<int32_t>(row_group_index), 0);
            if (!header_result) return header_result.error();
 
            thrift::CompactDecoder page_dec(header_result->data(), header_result->size());
            if (auto r = page_header.deserialize(page_dec); !r.has_value()) {
                return r.error();
            }
            if (page_dec.position() != header_result->size()) {
                return Error{ErrorCode::THRIFT_DECODE_ERROR,
                             "encrypted page header contains trailing bytes"};
            }
            header_size = 8 + static_cast<size_t>(encrypted_header_size);
        } else
#endif
        {
            // Deserialize the PageHeader to find where the data begins.
            // The serialized size of the PageHeader is variable (Thrift compact
            // encoding). We use decoder.position() to determine how many bytes
            // the PageHeader consumed.
            thrift::CompactDecoder page_dec(page_start, remaining);
            if (auto r = page_header.deserialize(page_dec); !r.has_value()) {
                return r.error();
            }
            header_size = page_dec.position();
        }
        if (page_header.compressed_page_size < 0) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "negative compressed_page_size"};
        }
        size_t page_data_size = static_cast<size_t>(
            page_header.compressed_page_size);
        const uint8_t* page_data = page_start + header_size;
 
        if (header_size + page_data_size > remaining) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "page data extends past end of file"};
        }
 
        // EX-1: Verify page CRC-32 if present in the page header.
        if (page_header.crc.has_value()) {
            uint32_t expected_crc = static_cast<uint32_t>(*page_header.crc);
            uint32_t computed_crc = detail_reader::crc32(page_data, page_data_size);
            if (computed_crc != expected_crc) {
                return Error{ErrorCode::CORRUPT_PAGE,
                             "Page CRC-32 mismatch at offset "
                             + std::to_string(offset) + ": expected 0x"
                             + ([&]{
                                    char buf[9]; std::snprintf(buf, sizeof(buf), "%08X", expected_crc);
                                    return std::string(buf);
                                })()
                             + ", computed 0x"
                             + ([&]{
                                    char buf[9]; std::snprintf(buf, sizeof(buf), "%08X", computed_crc);
                                    return std::string(buf);
                                })()};
            }
        }
 
        // --- Decrypt page data if PME is configured for this column ---
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
        if (decryptor_ && decryptor_has_column_key(decryptor_->config(), col_name)) {
            auto dec_result = (page_header.type == PageType::DICTIONARY_PAGE)
                ? decryptor_->decrypt_dict_page(
                      page_data, page_data_size, col_name,
                      static_cast<int32_t>(row_group_index))
                : decryptor_->decrypt_column_page(
                      page_data, page_data_size, col_name,
                      static_cast<int32_t>(row_group_index), 0);
            if (!dec_result) return dec_result.error();
            decompressed_buffers_.push_back(std::move(*dec_result));
            page_data = decompressed_buffers_.back().data();
            page_data_size = decompressed_buffers_.back().size();
        }
#endif
 
        // --- Decompress page data if needed ---
        if (col_meta.codec != Compression::UNCOMPRESSED) {
            size_t uncompressed_size = static_cast<size_t>(
                page_header.uncompressed_page_size);
            if (uncompressed_size == 0 || uncompressed_size > PARQUET_MAX_PAGE_SIZE) {
                return Error{ErrorCode::CORRUPT_PAGE,
                             "ParquetReader: uncompressed_page_size exceeds 256 MB hard cap"};
            }
            auto decompressed = decompress(col_meta.codec,
                                           page_data, page_data_size,
                                           uncompressed_size);
            if (!decompressed) {
                return Error{ErrorCode::UNSUPPORTED_COMPRESSION,
                             "decompression failed: " +
                             decompressed.error().message};
            }
            decompressed_buffers_.push_back(std::move(decompressed.value()));
            page_data = decompressed_buffers_.back().data();
            page_data_size = decompressed_buffers_.back().size();
        }
 
        // Determine num_values from the PageHeader
        int64_t num_values = 0;
        if (page_header.type == PageType::DATA_PAGE &&
            page_header.data_page_header.has_value()) {
            num_values = page_header.data_page_header->num_values;
        } else if (page_header.type == PageType::DATA_PAGE_V2 &&
                   page_header.data_page_header_v2.has_value()) {
            num_values = page_header.data_page_header_v2->num_values;
        } else {
            // Fall back to column metadata num_values
            num_values = col_meta.num_values;
        }
 
        if (num_values < 0 || num_values > MAX_VALUES_PER_PAGE) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "num_values out of valid range"};
        }
 
        // Determine the physical type and type_length from the schema
        PhysicalType pt = col_meta.type;
        int32_t type_length = -1;
        if (column_index < schema_.num_columns()) {
            type_length = schema_.column(column_index).type_length;
        }
 
        // M22: Check that decoded page won't exceed a reasonable memory budget
        {
            size_t elem_size = 0;
            switch (pt) {
                case PhysicalType::BOOLEAN: elem_size = 1; break;
                case PhysicalType::INT32:
                case PhysicalType::FLOAT:   elem_size = 4; break;
                case PhysicalType::INT64:
                case PhysicalType::DOUBLE:  elem_size = 8; break;
                case PhysicalType::FIXED_LEN_BYTE_ARRAY:
                    elem_size = (type_length > 0) ? static_cast<size_t>(type_length) : 1;
                    break;
                default: elem_size = 0; break; // variable-length: skip budget check
            }
            // CWE-400: Uncontrolled Resource Consumption — 256 MB decoded page memory budget
            if (elem_size > 0) {
                size_t decoded_size = static_cast<size_t>(num_values) * elem_size;
                if (decoded_size > 256ULL * 1024 * 1024) {
                    return Error{ErrorCode::CORRUPT_DATA,
                                 "decoded page exceeds 256 MB memory limit"};
                }
            }
        }
 
        ColumnReader col_reader(pt, page_data, page_data_size,
                                num_values, type_length);
 
        return ColumnReaderWithCount{std::move(col_reader), num_values};
    }
 
    struct PageReadResult {
        const uint8_t* data;        
        size_t         size;        
        thrift::PageHeader header;  
    };
 
    expected<PageReadResult> read_page_at(int64_t offset, Compression codec,
            const std::string& col_name = "",
            int32_t rg_index = 0, int32_t page_ordinal = 0) {
        if (offset < 0 || static_cast<size_t>(offset) >= file_data_.size()) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "page offset out of file bounds"};
        }
 
        size_t remaining = file_data_.size() - static_cast<size_t>(offset);
        const uint8_t* page_start = file_data_.data() + offset;
 
        thrift::PageHeader ph;
        size_t hdr_size = 0;
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
        const bool column_encrypted = decryptor_ && !col_name.empty() &&
            decryptor_has_column_key(decryptor_->config(), col_name);
        if (has_encrypted_page_header_prefix(page_start, remaining)) {
            if (!column_encrypted) {
                return Error{ErrorCode::ENCRYPTION_ERROR,
                             "encrypted page header encountered without matching decryptor for column '" +
                                 col_name + "'"};
            }
            const uint32_t encrypted_header_size = load_le32(page_start + 4);
            if (encrypted_header_size == 0 ||
                static_cast<size_t>(encrypted_header_size) > remaining - 8) {
                return Error{ErrorCode::CORRUPT_PAGE,
                             "encrypted page header size out of range"};
            }
            auto header_result = decryptor_->decrypt_data_page_header(
                page_start + 8, encrypted_header_size, col_name,
                rg_index, page_ordinal);
            if (!header_result) return header_result.error();
 
            thrift::CompactDecoder page_dec(header_result->data(), header_result->size());
            if (auto r = ph.deserialize(page_dec); !r.has_value()) {
                return r.error();
            }
            if (page_dec.position() != header_result->size()) {
                return Error{ErrorCode::THRIFT_DECODE_ERROR,
                             "encrypted page header contains trailing bytes"};
            }
            hdr_size = 8 + static_cast<size_t>(encrypted_header_size);
        } else
#endif
        {
            thrift::CompactDecoder page_dec(page_start, remaining);
            if (auto r = ph.deserialize(page_dec); !r.has_value()) {
                return r.error();
            }
            hdr_size = page_dec.position();
        }
        if (ph.compressed_page_size < 0) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "negative compressed_page_size"};
        }
        size_t compressed_size = static_cast<size_t>(ph.compressed_page_size);
        const uint8_t* pdata = page_start + hdr_size;
 
        if (hdr_size + compressed_size > remaining) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "page data extends past end of file"};
        }
 
        size_t pdata_size = compressed_size;
 
        // EX-1: Verify page CRC-32 if present in the page header.
        // Parquet spec field 4 is an optional CRC-32 over the compressed page data.
        if (ph.crc.has_value()) {
            uint32_t expected_crc = static_cast<uint32_t>(*ph.crc);
            uint32_t computed_crc = detail_reader::crc32(pdata, pdata_size);
            if (computed_crc != expected_crc) {
                return Error{ErrorCode::CORRUPT_PAGE,
                             "Page CRC-32 mismatch at offset "
                             + std::to_string(offset) + ": expected 0x"
                             + ([&]{
                                    char buf[9]; std::snprintf(buf, sizeof(buf), "%08X", expected_crc);
                                    return std::string(buf);
                                })()
                             + ", computed 0x"
                             + ([&]{
                                    char buf[9]; std::snprintf(buf, sizeof(buf), "%08X", computed_crc);
                                    return std::string(buf);
                                })()};
            }
        }
 
        // --- Decrypt page data if PME is configured ---
#if defined(SIGNET_ENABLE_COMMERCIAL) && SIGNET_ENABLE_COMMERCIAL
        if (decryptor_ && !col_name.empty() &&
            decryptor_has_column_key(decryptor_->config(), col_name)) {
            auto dec_result = (ph.type == PageType::DICTIONARY_PAGE)
                ? decryptor_->decrypt_dict_page(pdata, pdata_size, col_name, rg_index)
                : decryptor_->decrypt_column_page(pdata, pdata_size, col_name, rg_index, page_ordinal);
            if (!dec_result) return dec_result.error();
            decompressed_buffers_.push_back(std::move(*dec_result));
            pdata = decompressed_buffers_.back().data();
            pdata_size = decompressed_buffers_.back().size();
        }
#endif
 
        if (codec != Compression::UNCOMPRESSED) {
            size_t uncompressed_size = static_cast<size_t>(
                ph.uncompressed_page_size);
            if (uncompressed_size == 0 || uncompressed_size > PARQUET_MAX_PAGE_SIZE) {
                return Error{ErrorCode::CORRUPT_PAGE,
                             "ParquetReader: uncompressed_page_size exceeds 256 MB hard cap"};
            }
            // CWE-409: Improper Handling of Highly Compressed Data (Zip Bomb)
            // M20: Reject suspiciously high decompression ratios (zip bomb guard)
            if (pdata_size > 0 && uncompressed_size / pdata_size > 1024) {
                return Error{ErrorCode::CORRUPT_DATA,
                             "ParquetReader: decompression ratio exceeds 1024x limit"};
            }
            auto dec_result = decompress(codec, pdata, pdata_size,
                                         uncompressed_size);
            if (!dec_result) {
                return Error{ErrorCode::UNSUPPORTED_COMPRESSION,
                             "decompression failed: " +
                             dec_result.error().message};
            }
            decompressed_buffers_.push_back(std::move(dec_result.value()));
            pdata = decompressed_buffers_.back().data();
            pdata_size = decompressed_buffers_.back().size();
        }
 
        return PageReadResult{pdata, pdata_size, std::move(ph)};
    }
 
    template <typename T>
    expected<std::vector<T>> read_column_dict(
            const thrift::ColumnMetaData& col_meta,
            const std::string& col_name = "",
            int32_t rg_index = 0) {
        // Step 1: determine where the dictionary page lives
        int64_t dict_offset = col_meta.dictionary_page_offset.value_or(
            col_meta.data_page_offset);
 
        // Step 2: read the dictionary page
        auto dict_page_result = read_page_at(dict_offset, col_meta.codec,
                                              col_name, rg_index, -1);
        if (!dict_page_result) return dict_page_result.error();
 
        auto& dict_pr = dict_page_result.value();
        if (dict_pr.header.type != PageType::DICTIONARY_PAGE ||
            !dict_pr.header.dictionary_page_header.has_value()) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "expected DICTIONARY_PAGE at dictionary offset"};
        }
 
        int32_t raw_dict_count = dict_pr.header.dictionary_page_header->num_values;
        if (raw_dict_count < 0 || raw_dict_count > 10'000'000) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "dictionary page num_values out of valid range"};
        }
        size_t num_dict_entries = static_cast<size_t>(raw_dict_count);
 
        // Step 3: find the data page offset
        // If dictionary_page_offset was set and != data_page_offset, the
        // data page is at data_page_offset.  Otherwise the data page
        // immediately follows the dictionary page in the file.
        int64_t data_offset = col_meta.data_page_offset;
        if (data_offset == dict_offset) {
            // Dictionary page and data page are sequential; skip past the
            // dictionary page in the file to get to the data page.
            // We need the raw page size (compressed) + header.
            size_t dict_raw_start = static_cast<size_t>(dict_offset);
            const uint8_t* dict_start = file_data_.data() + dict_raw_start;
            size_t dict_remaining = file_data_.size() - dict_raw_start;
 
            thrift::CompactDecoder hdr_dec(dict_start, dict_remaining);
            thrift::PageHeader tmp_hdr;
            if (auto r = tmp_hdr.deserialize(hdr_dec); !r.has_value()) {
                return r.error();
            }
            size_t dict_hdr_size = hdr_dec.position();
            if (tmp_hdr.compressed_page_size < 0) {
                return Error{ErrorCode::CORRUPT_PAGE,
                             "negative compressed_page_size in dictionary page"};
            }
            size_t dict_compressed_size = static_cast<size_t>(
                tmp_hdr.compressed_page_size);
 
            data_offset = dict_offset
                        + static_cast<int64_t>(dict_hdr_size)
                        + static_cast<int64_t>(dict_compressed_size);
        }
 
        // Step 4: read the data page (contains RLE-encoded indices)
        auto data_page_result = read_page_at(data_offset, col_meta.codec,
                                              col_name, rg_index, 0);
        if (!data_page_result) return data_page_result.error();
 
        auto& data_pr = data_page_result.value();
 
        int64_t num_values = 0;
        if (data_pr.header.type == PageType::DATA_PAGE &&
            data_pr.header.data_page_header.has_value()) {
            num_values = data_pr.header.data_page_header->num_values;
        } else if (data_pr.header.type == PageType::DATA_PAGE_V2 &&
                   data_pr.header.data_page_header_v2.has_value()) {
            num_values = data_pr.header.data_page_header_v2->num_values;
        } else {
            num_values = col_meta.num_values;
        }
 
        if (num_values < 0 || num_values > MAX_VALUES_PER_PAGE) {
            return Error{ErrorCode::CORRUPT_PAGE,
                         "num_values out of valid range"};
        }
 
        // Step 5: decode using DictionaryDecoder
        DictionaryDecoder<T> decoder(dict_pr.data, dict_pr.size,
                                     num_dict_entries, col_meta.type);
 
        return decoder.decode(data_pr.data, data_pr.size,
                              static_cast<size_t>(num_values));
    }
 
    template <typename T>
    expected<std::vector<T>> read_column_delta(
            const thrift::ColumnMetaData& col_meta,
            const std::string& col_name = "",
            int32_t rg_index = 0) {
        auto page_result = read_page_at(col_meta.data_page_offset,
                                        col_meta.codec,
                                        col_name, rg_index, 0);
        if (!page_result) return page_result.error();
 
        auto& pr = page_result.value();
 
        int64_t num_values = 0;
        if (pr.header.type == PageType::DATA_PAGE &&
            pr.header.data_page_header.has_value()) {
            num_values = pr.header.data_page_header->num_values;
        } else if (pr.header.type == PageType::DATA_PAGE_V2 &&
                   pr.header.data_page_header_v2.has_value()) {
            num_values = pr.header.data_page_header_v2->num_values;
        } else {
            num_values = col_meta.num_values;
        }
 
        auto count_result = validate_page_value_count(num_values, "ParquetReader DELTA_BINARY_PACKED");
        if (!count_result) return count_result.error();
        size_t count = *count_result;
 
        if constexpr (std::is_same_v<T, int32_t>) {
            return delta::decode_int32(pr.data, pr.size, count);
        } else if constexpr (std::is_same_v<T, int64_t>) {
            return delta::decode_int64(pr.data, pr.size, count);
        } else {
            return Error{ErrorCode::UNSUPPORTED_ENCODING,
                         "DELTA_BINARY_PACKED only supports INT32/INT64"};
        }
    }
 
    template <typename T>
    expected<std::vector<T>> read_column_bss(
            const thrift::ColumnMetaData& col_meta,
            const std::string& col_name = "",
            int32_t rg_index = 0) {
        auto page_result = read_page_at(col_meta.data_page_offset,
                                        col_meta.codec,
                                        col_name, rg_index, 0);
        if (!page_result) return page_result.error();
 
        auto& pr = page_result.value();
 
        int64_t num_values = 0;
        if (pr.header.type == PageType::DATA_PAGE &&
            pr.header.data_page_header.has_value()) {
            num_values = pr.header.data_page_header->num_values;
        } else if (pr.header.type == PageType::DATA_PAGE_V2 &&
                   pr.header.data_page_header_v2.has_value()) {
            num_values = pr.header.data_page_header_v2->num_values;
        } else {
            num_values = col_meta.num_values;
        }
 
        auto count_result = validate_page_value_count(num_values, "ParquetReader BYTE_STREAM_SPLIT");
        if (!count_result) return count_result.error();
        size_t count = *count_result;
 
        if constexpr (std::is_same_v<T, float>) {
            return byte_stream_split::decode_float(pr.data, pr.size, count);
        } else if constexpr (std::is_same_v<T, double>) {
            return byte_stream_split::decode_double(pr.data, pr.size, count);
        } else {
            return Error{ErrorCode::UNSUPPORTED_ENCODING,
                         "BYTE_STREAM_SPLIT only supports FLOAT/DOUBLE"};
        }
    }
 
    template <typename T>
    expected<std::vector<T>> read_column_rle_bool(
            const thrift::ColumnMetaData& col_meta,
            const std::string& col_name = "",
            int32_t rg_index = 0) {
        auto page_result = read_page_at(col_meta.data_page_offset,
                                        col_meta.codec,
                                        col_name, rg_index, 0);
        if (!page_result) return page_result.error();
 
        auto& pr = page_result.value();
 
        int64_t num_values = 0;
        if (pr.header.type == PageType::DATA_PAGE &&
            pr.header.data_page_header.has_value()) {
            num_values = pr.header.data_page_header->num_values;
        } else if (pr.header.type == PageType::DATA_PAGE_V2 &&
                   pr.header.data_page_header_v2.has_value()) {
            num_values = pr.header.data_page_header_v2->num_values;
        } else {
            num_values = col_meta.num_values;
        }
 
        size_t count = static_cast<size_t>(num_values);
 
        if constexpr (std::is_same_v<T, bool>) {
            // RLE boolean: 4-byte LE length prefix + RLE payload, bit_width=1
            auto indices = RleDecoder::decode_with_length(
                pr.data, pr.size, /*bit_width=*/1, count);
 
            std::vector<bool> result;
            result.reserve(count);
            for (size_t i = 0; i < count && i < indices.size(); ++i) {
                result.push_back(indices[i] != 0);
            }
            return result;
        } else {
            return Error{ErrorCode::UNSUPPORTED_ENCODING,
                         "RLE encoding for booleans requires bool type"};
        }
    }
 
    static std::vector<std::string> to_string_vec(const std::vector<bool>& vals) {
        std::vector<std::string> result;
        result.reserve(vals.size());
        for (bool v : vals) {
            result.push_back(v ? "true" : "false");
        }
        return result;
    }
 
    template <typename T>
    static std::vector<std::string> to_string_vec(const std::vector<T>& vals) {
        std::vector<std::string> result;
        result.reserve(vals.size());
        for (const auto& v : vals) {
            result.push_back(std::to_string(v));
        }
        return result;
    }
 
    static std::string hex_encode(const std::vector<uint8_t>& bytes) {
        static constexpr char hex_chars[] = "0123456789abcdef";
        std::string result;
        result.reserve(bytes.size() * 2);
        for (uint8_t b : bytes) {
            result.push_back(hex_chars[(b >> 4) & 0x0F]);
            result.push_back(hex_chars[b & 0x0F]);
        }
        return result;
    }
 
    static LogicalType converted_type_to_logical(ConvertedType ct) {
        switch (ct) {
        case ConvertedType::UTF8:             return LogicalType::STRING;
        case ConvertedType::ENUM:             return LogicalType::ENUM;
        case ConvertedType::DATE:             return LogicalType::DATE;
        case ConvertedType::TIME_MILLIS:      return LogicalType::TIME_MS;
        case ConvertedType::TIME_MICROS:      return LogicalType::TIME_US;
        case ConvertedType::TIMESTAMP_MILLIS: return LogicalType::TIMESTAMP_MS;
        case ConvertedType::TIMESTAMP_MICROS: return LogicalType::TIMESTAMP_US;
        case ConvertedType::DECIMAL:          return LogicalType::DECIMAL;
        case ConvertedType::JSON:             return LogicalType::JSON;
        case ConvertedType::BSON:             return LogicalType::BSON;
        default:                              return LogicalType::NONE;
        }
    }
};
 
} // namespace signet::forge