From 8ebd267d5ab9bdeea0d4e3565a38b310d7a00ab0 Mon Sep 17 00:00:00 2001 From: Alejandro Valerio Date: Thu, 5 Mar 2026 14:28:04 -0600 Subject: [PATCH] - Organizing parquet_utils.cpp and parquet_utils.h into sections before rewrite for adapter version. --- src/processing/parquet_utils.cpp | 186 ++++++++++++++++--------------- src/processing/parquet_utils.h | 40 +++---- 2 files changed, 118 insertions(+), 108 deletions(-) diff --git a/src/processing/parquet_utils.cpp b/src/processing/parquet_utils.cpp index f1660ed..2309fa2 100644 --- a/src/processing/parquet_utils.cpp +++ b/src/processing/parquet_utils.cpp @@ -25,6 +25,10 @@ using namespace dbps::external; using namespace dbps::enum_utils; using namespace dbps::compression; +// ----------------------------------------------------------------------------- +// Process Parquet formatted Dictionary and Data pages +// ----------------------------------------------------------------------------- + int CalculateLevelBytesLength(tcb::span raw, const AttributesMap& encoding_attribs) { @@ -101,6 +105,98 @@ int CalculateLevelBytesLength(tcb::span raw, return total_level_bytes; } +LevelAndValueBytes DecompressAndSplit( + tcb::span plaintext, + CompressionCodec::type compression, + const AttributesMap& encoding_attributes) { + + // Get the page type from the encoding attributes. + auto page_type = std::get(encoding_attributes.at("page_type")); + + // On DATA_PAGE_V1, the whole payload is compressed. + // So the split of level and value byte requires to + // (1) decompress the whole payload, (2) calculate length of level bytes, (3) split into level and value bytes. + if (page_type == "DATA_PAGE_V1") { + auto decompressed_bytes = Decompress(plaintext, compression); + int leading_bytes_to_strip = CalculateLevelBytesLength( + decompressed_bytes, encoding_attributes); + auto [level_bytes, value_bytes] = Split(decompressed_bytes, leading_bytes_to_strip); + return LevelAndValueBytes{std::move(level_bytes), std::move(value_bytes)}; + } + + // On DATA_PAGE_V2, only the value bytes are compressed. + // So the split of level and value byte requires to + // (1) calculate length of level bytes, (2) split into level, (3) decompress only the value bytes. + if (page_type == "DATA_PAGE_V2") { + int leading_bytes_to_strip = CalculateLevelBytesLength( + plaintext, encoding_attributes); + auto [level_bytes_span, compressed_value_bytes_span] = Split(plaintext, leading_bytes_to_strip); + std::vector level_bytes(level_bytes_span.begin(), level_bytes_span.end()); + + bool page_v2_is_compressed = std::get( + encoding_attributes.at("page_v2_is_compressed")); + std::vector value_bytes; + if (page_v2_is_compressed) { + value_bytes = Decompress(compressed_value_bytes_span, compression); + } else { + value_bytes = std::vector(compressed_value_bytes_span.begin(), compressed_value_bytes_span.end()); + } + return LevelAndValueBytes{std::move(level_bytes), std::move(value_bytes)}; + } + + // DICTIONARY_PAGE has no level bytes. + if (page_type == "DICTIONARY_PAGE") { + auto level_bytes = std::vector(); + auto value_bytes = Decompress(plaintext, compression); + return LevelAndValueBytes{std::move(level_bytes), std::move(value_bytes)}; + } + + throw InvalidInputException("Unexpected page type: " + page_type); +} + +std::vector CompressAndJoin( + const std::vector& level_bytes, + const std::vector& value_bytes, + CompressionCodec::type compression, + const AttributesMap& encoding_attributes) { + + // Get the page type from the encoding attributes. + const auto& page_type = std::get(encoding_attributes.at("page_type")); + + // Check that the calculated level bytes size == the size of the actual level bytes. + int expected_level_bytes = CalculateLevelBytesLength(level_bytes, encoding_attributes); + if (static_cast(expected_level_bytes) != level_bytes.size()) { + throw InvalidInputException("Level bytes size does not match encoding attributes"); + } + + if (page_type == "DATA_PAGE_V1") { + auto joined = Join(level_bytes, value_bytes); + return Compress(joined, compression); + } + + if (page_type == "DATA_PAGE_V2") { + bool page_v2_is_compressed = + std::get(encoding_attributes.at("page_v2_is_compressed")); + if (page_v2_is_compressed) { + auto compressed_values = Compress(value_bytes, compression); + return Join(level_bytes, compressed_values); + } else { + return Join(level_bytes, value_bytes); + } + } + + // DICTIONARY_PAGE has no level bytes. + if (page_type == "DICTIONARY_PAGE") { + return Compress(value_bytes, compression); + } + + throw InvalidInputException("Unexpected page type: " + page_type); +} + +// ----------------------------------------------------------------------------- +// Parse and build Parquet formatted value bytes into vectors -- version materialized-vectors +// ----------------------------------------------------------------------------- + inline static size_t GetFixedElemSizeOrThrow(Type::type datatype, const std::optional& datatype_length) { switch (datatype) { case Type::INT32: @@ -261,94 +357,6 @@ std::vector ParseByteArrayListValueBytes(const std::vector return std::move(*values); } -LevelAndValueBytes DecompressAndSplit( - tcb::span plaintext, - CompressionCodec::type compression, - const AttributesMap& encoding_attributes) { - - // Get the page type from the encoding attributes. - auto page_type = std::get(encoding_attributes.at("page_type")); - - // On DATA_PAGE_V1, the whole payload is compressed. - // So the split of level and value byte requires to - // (1) decompress the whole payload, (2) calculate length of level bytes, (3) split into level and value bytes. - if (page_type == "DATA_PAGE_V1") { - auto decompressed_bytes = Decompress(plaintext, compression); - int leading_bytes_to_strip = CalculateLevelBytesLength( - decompressed_bytes, encoding_attributes); - auto [level_bytes, value_bytes] = Split(decompressed_bytes, leading_bytes_to_strip); - return LevelAndValueBytes{std::move(level_bytes), std::move(value_bytes)}; - } - - // On DATA_PAGE_V2, only the value bytes are compressed. - // So the split of level and value byte requires to - // (1) calculate length of level bytes, (2) split into level, (3) decompress only the value bytes. - if (page_type == "DATA_PAGE_V2") { - int leading_bytes_to_strip = CalculateLevelBytesLength( - plaintext, encoding_attributes); - auto [level_bytes_span, compressed_value_bytes_span] = Split(plaintext, leading_bytes_to_strip); - std::vector level_bytes(level_bytes_span.begin(), level_bytes_span.end()); - - bool page_v2_is_compressed = std::get( - encoding_attributes.at("page_v2_is_compressed")); - std::vector value_bytes; - if (page_v2_is_compressed) { - value_bytes = Decompress(compressed_value_bytes_span, compression); - } else { - value_bytes = std::vector(compressed_value_bytes_span.begin(), compressed_value_bytes_span.end()); - } - return LevelAndValueBytes{std::move(level_bytes), std::move(value_bytes)}; - } - - // DICTIONARY_PAGE has no level bytes. - if (page_type == "DICTIONARY_PAGE") { - auto level_bytes = std::vector(); - auto value_bytes = Decompress(plaintext, compression); - return LevelAndValueBytes{std::move(level_bytes), std::move(value_bytes)}; - } - - throw InvalidInputException("Unexpected page type: " + page_type); -} - -std::vector CompressAndJoin( - const std::vector& level_bytes, - const std::vector& value_bytes, - CompressionCodec::type compression, - const AttributesMap& encoding_attributes) { - - // Get the page type from the encoding attributes. - const auto& page_type = std::get(encoding_attributes.at("page_type")); - - // Check that the calculated level bytes size == the size of the actual level bytes. - int expected_level_bytes = CalculateLevelBytesLength(level_bytes, encoding_attributes); - if (static_cast(expected_level_bytes) != level_bytes.size()) { - throw InvalidInputException("Level bytes size does not match encoding attributes"); - } - - if (page_type == "DATA_PAGE_V1") { - auto joined = Join(level_bytes, value_bytes); - return Compress(joined, compression); - } - - if (page_type == "DATA_PAGE_V2") { - bool page_v2_is_compressed = - std::get(encoding_attributes.at("page_v2_is_compressed")); - if (page_v2_is_compressed) { - auto compressed_values = Compress(value_bytes, compression); - return Join(level_bytes, compressed_values); - } else { - return Join(level_bytes, value_bytes); - } - } - - // DICTIONARY_PAGE has no level bytes. - if (page_type == "DICTIONARY_PAGE") { - return Compress(value_bytes, compression); - } - - throw InvalidInputException("Unexpected page type: " + page_type); -} - TypedListValues ParseValueBytesIntoTypedList( const std::vector& bytes, Type::type datatype, @@ -367,3 +375,5 @@ std::vector GetTypedListAsValueBytes( std::vector raw_values = BuildRawBytesFromTypedListValues(list); return CombineRawBytesIntoValueBytes(raw_values, datatype, datatype_length, encoding); } + +// ----------------------------------------------------------------------------- diff --git a/src/processing/parquet_utils.h b/src/processing/parquet_utils.h index 7c8e6dc..45da615 100644 --- a/src/processing/parquet_utils.h +++ b/src/processing/parquet_utils.h @@ -48,6 +48,26 @@ using namespace dbps::external; int CalculateLevelBytesLength(tcb::span raw, const AttributesMap& encoding_attribs); +/** + * Decompresses and splits a Parquet page into level and value bytes. + * Handles DATA_PAGE_V1, DATA_PAGE_V2 (including optional compression on value bytes), + * and DICTIONARY_PAGE. + */ + LevelAndValueBytes DecompressAndSplit( + tcb::span plaintext, + CompressionCodec::type compression, + const AttributesMap& encoding_attributes); + +/** + * Reverse of DecompressAndSplit: joins level/value bytes and applies compression + * based on page type and encoding attributes. + */ +std::vector CompressAndJoin( + const std::vector& level_bytes, + const std::vector& value_bytes, + CompressionCodec::type compression, + const AttributesMap& encoding_attributes); + /** * Slice a flat byte buffer into RawValueBytes elements according to datatype/encoding. * This follows the Parquet specific encoding. @@ -77,26 +97,6 @@ std::vector BuildByteArrayValueBytes(const std::string& payload); */ std::vector ParseByteArrayListValueBytes(const std::vector& bytes); -/** - * Decompresses and splits a Parquet page into level and value bytes. - * Handles DATA_PAGE_V1, DATA_PAGE_V2 (including optional compression on value bytes), - * and DICTIONARY_PAGE. - */ -LevelAndValueBytes DecompressAndSplit( - tcb::span plaintext, - CompressionCodec::type compression, - const AttributesMap& encoding_attributes); - -/** - * Reverse of DecompressAndSplit: joins level/value bytes and applies compression - * based on page type and encoding attributes. - */ -std::vector CompressAndJoin( - const std::vector& level_bytes, - const std::vector& value_bytes, - CompressionCodec::type compression, - const AttributesMap& encoding_attributes); - /** * Parse the value bytes into a typed list based on the data type and encoding. *