column_writer.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright 2026 Johnson Ogundeji
#pragma once
 
 
#include "signet/types.hpp"
#include "signet/statistics.hpp"
 
#include <cstdint>
#include <cstring>
#include <stdexcept>
#include <string>
#include <type_traits>
#include <vector>
 
namespace signet::forge {
 
// ---------------------------------------------------------------------------
// Little-endian append helpers
// ---------------------------------------------------------------------------
 
inline void append_le32(std::vector<uint8_t>& buf, uint32_t val) {
    buf.push_back(static_cast<uint8_t>((val      ) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >>  8) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 16) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 24) & 0xFF));
}
 
inline void append_le64(std::vector<uint8_t>& buf, uint64_t val) {
    buf.push_back(static_cast<uint8_t>((val      ) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >>  8) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 16) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 24) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 32) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 40) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 48) & 0xFF));
    buf.push_back(static_cast<uint8_t>((val >> 56) & 0xFF));
}
 
class ColumnWriter {
public:
    explicit ColumnWriter(PhysicalType type, int32_t type_length = -1)
        : type_(type), type_length_(type_length), num_values_(0) {}
 
    // -- Typed write methods --------------------------------------------------
 
    void write_bool(bool val) {
        size_t bit_index = static_cast<size_t>(num_values_);
        size_t byte_index = bit_index / 8;
        size_t bit_offset = bit_index % 8;
 
        // Extend the buffer if we need a new byte
        if (byte_index >= buf_.size()) {
            buf_.push_back(0);
        }
 
        if (val) {
            buf_[byte_index] |= static_cast<uint8_t>(1u << bit_offset);
        }
 
        stats_.update(val);
        ++num_values_;
    }
 
    void write_int32(int32_t val) {
        uint32_t bits;
        std::memcpy(&bits, &val, sizeof(bits));
        append_le32(buf_, bits);
 
        stats_.update(val);
        ++num_values_;
    }
 
    void write_int64(int64_t val) {
        uint64_t bits;
        std::memcpy(&bits, &val, sizeof(bits));
        append_le64(buf_, bits);
 
        stats_.update(val);
        ++num_values_;
    }
 
    void write_float(float val) {
        uint32_t bits;
        std::memcpy(&bits, &val, sizeof(bits));
        append_le32(buf_, bits);
 
        stats_.update(val);
        ++num_values_;
    }
 
    void write_double(double val) {
        uint64_t bits;
        std::memcpy(&bits, &val, sizeof(bits));
        append_le64(buf_, bits);
 
        stats_.update(val);
        ++num_values_;
    }
 
    void write_byte_array(const std::string& val) {
        write_byte_array(reinterpret_cast<const uint8_t*>(val.data()), val.size());
    }
 
    void write_byte_array(const uint8_t* data, size_t len) {
        // CWE-190: Integer Overflow — Parquet BYTE_ARRAY uses 4-byte LE length prefix;
        // payloads > 4 GiB would truncate the length field, corrupting the output.
        if (len > static_cast<size_t>(UINT32_MAX)) {
            // H14: BYTE_ARRAY length prefix is a 4-byte LE uint32; reject > 4 GiB payloads.
            // Throw instead of silently dropping the value, which would corrupt the output.
            throw std::length_error("BYTE_ARRAY value exceeds 4 GiB limit");
        }
        append_le32(buf_, static_cast<uint32_t>(len));
        buf_.insert(buf_.end(), data, data + len);
 
        // Update statistics with the string representation
        std::string str_val(reinterpret_cast<const char*>(data), len);
        stats_.update(str_val);
        ++num_values_;
    }
 
    void write_fixed_len_byte_array(const uint8_t* data, size_t len) {
        // CWE-130: Improper Handling of Length Parameter Inconsistency —
        // Validate that len matches the schema's type_length to prevent
        // silent data corruption in the column chunk.
        if (type_length_ > 0 && len != static_cast<size_t>(type_length_)) {
            throw std::length_error(
                "FIXED_LEN_BYTE_ARRAY value length " + std::to_string(len) +
                " != schema type_length " + std::to_string(type_length_));
        }
        buf_.insert(buf_.end(), data, data + len);
 
        // Update statistics with the string representation
        std::string str_val(reinterpret_cast<const char*>(data), len);
        stats_.update(str_val);
        ++num_values_;
    }
 
    // -- Template convenience dispatchers ------------------------------------
 
    template <typename T>
    void write(const T& val) {
        if constexpr (std::is_same_v<T, bool>) {
            write_bool(val);
        } else if constexpr (std::is_same_v<T, int32_t>) {
            write_int32(val);
        } else if constexpr (std::is_same_v<T, int64_t>) {
            write_int64(val);
        } else if constexpr (std::is_same_v<T, float>) {
            write_float(val);
        } else if constexpr (std::is_same_v<T, double>) {
            write_double(val);
        } else if constexpr (std::is_same_v<T, std::string>) {
            write_byte_array(val);
        } else {
            static_assert(!std::is_same_v<T, T>,
                          "ColumnWriter::write: unsupported type");
        }
    }
 
    // -- Batch write methods -------------------------------------------------
 
    template <typename T>
    void write_batch(const T* values, size_t count) {
        for (size_t i = 0; i < count; ++i) {
            write<T>(values[i]);
        }
    }
 
    void write_batch(const std::string* values, size_t count) {
        for (size_t i = 0; i < count; ++i) {
            write_byte_array(values[i]);
        }
    }
 
    // -- Access encoded data -------------------------------------------------
 
    [[nodiscard]] const std::vector<uint8_t>& data() const { return buf_; }
 
    [[nodiscard]] size_t encoded_size() const { return buf_.size(); }
 
    [[nodiscard]] int64_t num_values() const { return num_values_; }
 
    // -- Statistics -----------------------------------------------------------
 
    [[nodiscard]] const ColumnStatistics& statistics() const { return stats_; }
 
    // -- Reset ----------------------------------------------------------------
 
    void reset() {
        buf_.clear();
        stats_.reset();
        num_values_ = 0;
    }
 
    [[nodiscard]] PhysicalType type() const { return type_; }
 
private:
    PhysicalType           type_;
    int32_t                type_length_;  
    std::vector<uint8_t>   buf_;
    ColumnStatistics       stats_;
    int64_t                num_values_;
};
 
} // namespace signet::forge