- Propagate num_elements Parquet-based to TypedValuesBuffer and BasicXorEncryptor.

- Now num_elements is an invariant / constant all along TypedValuesBuffer and BasicXorEncryptor (Yay!)
- Fixed issue on iterator to account for prefix size on input buffers.
- Many unittest fixes for interfaces updates for num_elements.

Author: avalerio-tkd

src/processing/encryption_sequencer.cpp

@@ -181,7 +181,8 @@ bool DataBatchEncryptionSequencer::DecodeAndEncrypt(tcb::span<const uint8_t> pla
 
         // Parse value bytes into typed values buffer
         stage_start = std::chrono::steady_clock::now();
-        auto typed_buffer = ReinterpretValueBytesAsTypedValuesBuffer(value_bytes, datatype_, datatype_length_, encoding_);
+        auto typed_buffer = ReinterpretValueBytesAsTypedValuesBuffer(
+            value_bytes, num_elements, datatype_, datatype_length_, encoding_);
         parse_value_bytes_into_typed_list_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
             std::chrono::steady_clock::now() - stage_start).count();
 

src/processing/encryption_sequencer_test.cpp

@@ -484,7 +484,7 @@ TEST(EncryptionSequencer, FixedLenByteArrayValidation) {
     EXPECT_TRUE(testValidationFailure(0, "FIXED_LEN_BYTE_ARRAY datatype_length must be positive"));
 
     // Test valid case (should pass parameter validation)
-    DataBatchEncryptionSequencer sequencer("test_column", Type::FIXED_LEN_BYTE_ARRAY, 16, CompressionCodec::UNCOMPRESSED, Encoding::PLAIN, {{"page_type", "DICTIONARY_PAGE"}, {"dict_page_num_values", "1"}}, CompressionCodec::UNCOMPRESSED, "test_key_123", "test_user", "{}", {});
+    DataBatchEncryptionSequencer sequencer("test_column", Type::FIXED_LEN_BYTE_ARRAY, 16, CompressionCodec::UNCOMPRESSED, Encoding::PLAIN, {{"page_type", "DICTIONARY_PAGE"}, {"dict_page_num_values", "2"}}, CompressionCodec::UNCOMPRESSED, "test_key_123", "test_user", "{}", {});
     bool result = sequencer.DecodeAndEncrypt(FIXED_LEN_BYTE_ARRAY_DATA);
 
     if (!result && sequencer.error_stage_ == "parameter_validation") {

src/processing/encryptors/basic_xor_encryptor.cpp

@@ -196,7 +196,7 @@ std::vector<uint8_t> BasicXorEncryptor::EncryptTypedElements(
     constexpr bool is_fixed = TypedBuffer::is_fixed_sized;
     constexpr size_t prefix_length = is_fixed ? kFixedHeaderLength : kVariableHeaderLength;
 
-    // const size_t num_elements = 500000;     // +++++++++ input_buffer.GetNumElements();
+    // GetNumElements is read from Parquet metadata and level bytes, not calculated from payload.
     const size_t num_elements = input_buffer.GetNumElements();
 
     // If there are no elements, return an empty buffer with the header.
@@ -229,15 +229,11 @@ std::vector<uint8_t> BasicXorEncryptor::EncryptTypedElements(
             num_elements, prefix_length, RawBytesFixedSizedCodec{element_size}};
 
         auto loop_start = std::chrono::steady_clock::now();
+
         size_t output_index = 0;
+        tcb::span<const uint8_t> raw_bytes;
 
-        // ++++++ Restore the while (auto rawbytes = input_buffer.ElementsIteratorNext()) {} form.
-        while (true) {
-            // auto t0 = std::chrono::steady_clock::now();
-            tcb::span<const uint8_t> raw_bytes;
-            if (!input_buffer.ElementsIteratorNext(raw_bytes)) break;
-            // get_raw_element_ns += ElapsedNanosecondsSince(t0);
-
+        while (input_buffer.ElementsIteratorNext(raw_bytes)) {
             // auto t1 = std::chrono::steady_clock::now();
             auto write_span = output_buffer.GetWritableRawElement(output_index, raw_bytes.size());
             // get_writable_element_ns += ElapsedNanosecondsSince(t1);
@@ -262,15 +258,14 @@ std::vector<uint8_t> BasicXorEncryptor::EncryptTypedElements(
             num_elements, reserved_bytes_hint, true, prefix_length};
 
         auto loop_start = std::chrono::steady_clock::now();
-        size_t output_index = 0;
 
-        // ++++++ Restore the while (auto rawbytes = input_buffer.ElementsIteratorNext()) {} form.
-        while (true) {
-            // auto t0 = std::chrono::steady_clock::now();
-            tcb::span<const uint8_t> raw_bytes;
-            if (!input_buffer.ElementsIteratorNext(raw_bytes)) break;
-            // get_raw_element_ns += ElapsedNanosecondsSince(t0);
+        size_t output_index = 0;
+        tcb::span<const uint8_t> raw_bytes;
 
+        // ++++++ REMOVE ALL EXTRA INSTRUMENTATION AND TIMING CODE.
+        // - CAREFUL WITH NOT REMOVING CORRECT INLINED COMMENTS !!!!
+        
+        while (input_buffer.ElementsIteratorNext(raw_bytes)) {
             // auto t1 = std::chrono::steady_clock::now();
             auto write_span = output_buffer.GetWritableRawElement(output_index, raw_bytes.size());
             // get_writable_element_ns += ElapsedNanosecondsSince(t1);
@@ -344,12 +339,11 @@ template <typename TypedBuffer>
 TypedBuffer BasicXorEncryptor::DecryptFixedSizedElementsIntoTypedBuffer(
     const TypedBufferRawBytesFixedSized& encrypted_buffer, TypedBuffer output_buffer) {
     size_t output_index = 0;
-    tcb::span<const uint8_t> raw_bytes;
-    for (auto raw_bytes : encrypted_buffer.raw_elements()) {
-    // +++++++ new iterator not working?
-    // +++++++ while (encrypted_buffer.ElementsIteratorNext(raw_bytes)) {
-        auto write_span = output_buffer.GetWritableRawElement(output_index, raw_bytes.size());
-        XorDecryptInto(raw_bytes, write_span);
+    tcb::span<const uint8_t> element_bytes;
+    size_t element_size = encrypted_buffer.GetElementSize();
+    while (encrypted_buffer.ElementsIteratorNext(element_bytes)) {
+        auto write_span = output_buffer.GetWritableRawElement(output_index, element_size);
+        XorDecryptInto(element_bytes, write_span);
         output_index++;
     }
     return output_buffer;
@@ -366,7 +360,7 @@ TypedValuesBuffer BasicXorEncryptor::DecryptValueList(
 
         // Create a fixed-sized byte buffer for reading the encrypted elements.
         TypedBufferRawBytesFixedSized encrypted_buffer{
-            encrypted_bytes, kFixedHeaderLength, RawBytesFixedSizedCodec{header.element_size}};
+            encrypted_bytes, num_elements, kFixedHeaderLength, RawBytesFixedSizedCodec{header.element_size}};
 
         // Populate a typed buffer with the decrypted elements in the corresponding type.
         switch (datatype_) {
@@ -399,21 +393,18 @@ TypedValuesBuffer BasicXorEncryptor::DecryptValueList(
     // Decrypt variable-size elements
     else {
         // Create a variable-sized byte buffer for reading the encrypted elements.
-        TypedBufferRawBytesVariableSized encrypted_buffer{ encrypted_bytes, kVariableHeaderLength};
+        TypedBufferRawBytesVariableSized encrypted_buffer{ encrypted_bytes, num_elements, kVariableHeaderLength};
 
         switch (datatype_) {
             // Create a BYTE-ARRAY typed buffer for storing the decrypted elements.
             case Type::BYTE_ARRAY: {
                 auto reserved_bytes_hint = encrypted_buffer.GetRawBufferSize();
                 TypedBufferRawBytesVariableSized output_buffer{num_elements, reserved_bytes_hint, true};
-
                 size_t output_index = 0;
-                tcb::span<const uint8_t> element;
-                for (auto element : encrypted_buffer.raw_elements()) {
-                // +++++++ new iterator not working?
-                // +++++++ while (encrypted_buffer.ElementsIteratorNext(element)) {
-                    auto write_span = output_buffer.GetWritableRawElement(output_index, element.size());
-                    XorDecryptInto(element, write_span);
+                tcb::span<const uint8_t> element_bytes;
+                while (encrypted_buffer.ElementsIteratorNext(element_bytes)) {
+                    auto write_span = output_buffer.GetWritableRawElement(output_index, element_bytes.size());
+                    XorDecryptInto(element_bytes, write_span);
                     output_index++;
                 }   
                 return output_buffer;

src/processing/encryptors/basic_xor_encryptor_test.cpp

@@ -71,7 +71,7 @@ TEST(BasicXorEncryptor, EncryptDecryptValueList_RoundTrip_INT32) {
     // re-wrap bytes as a read buffer to match production read path behavior.
     std::vector<uint8_t> input_buffer_bytes = input_buffer_write.FinalizeAndTakeBuffer();
     const auto input_span = tcb::span<const uint8_t>(input_buffer_bytes.data(), input_buffer_bytes.size());
-    TypedBufferI32 input_buffer_read{input_span};
+    TypedBufferI32 input_buffer_read{input_span, values.size()};
     TypedValuesBuffer typed_buffer = std::move(input_buffer_read);
 
     std::vector<uint8_t> encrypted_blob = encryptor.EncryptValueList(typed_buffer);
@@ -99,7 +99,7 @@ TEST(BasicXorEncryptor, EncryptDecryptValueList_RoundTrip_DOUBLE) {
     // re-wrap bytes as a read buffer to match production read path behavior.
     std::vector<uint8_t> input_buffer_bytes = input_buffer_write.FinalizeAndTakeBuffer();
     const auto input_span = tcb::span<const uint8_t>(input_buffer_bytes.data(), input_buffer_bytes.size());
-    TypedBufferDouble input_buffer_read{input_span};
+    TypedBufferDouble input_buffer_read{input_span, values.size()};
     TypedValuesBuffer typed_buffer = std::move(input_buffer_read);
 
     std::vector<uint8_t> encrypted_blob = encryptor.EncryptValueList(typed_buffer);
@@ -149,7 +149,7 @@ TEST(BasicXorEncryptor, EncryptDecryptValueList_RoundTrip_BYTE_ARRAY) {
     // re-wrap bytes as a read buffer to match production read path behavior.
     std::vector<uint8_t> input_buffer_bytes = input_buffer_write.FinalizeAndTakeBuffer();
     const auto input_span = tcb::span<const uint8_t>(input_buffer_bytes.data(), input_buffer_bytes.size());
-    TypedBufferRawBytesVariableSized input_buffer_read{input_span};
+    TypedBufferRawBytesVariableSized input_buffer_read{input_span, values.size()};
     TypedValuesBuffer typed_buffer = std::move(input_buffer_read);
 
     std::vector<uint8_t> encrypted_blob = encryptor.EncryptValueList(typed_buffer);

src/processing/parquet_utils.cpp

@@ -390,7 +390,9 @@ std::vector<uint8_t> CompressAndJoin(
 // Public functions to build Parquet formatted value bytes into TypedValuesBuffer
 // -----------------------------------------------------------------------------
 
-TypedValuesBuffer ReinterpretValueBytesAsTypedValuesBuffer(tcb::span<const uint8_t> value_bytes,
+TypedValuesBuffer ReinterpretValueBytesAsTypedValuesBuffer(
+    tcb::span<const uint8_t> value_bytes,
+    size_t num_elements,
     Type::type datatype,
     const std::optional<int>& datatype_length,
     Encoding::type encoding) {
@@ -414,24 +416,24 @@ TypedValuesBuffer ReinterpretValueBytesAsTypedValuesBuffer(tcb::span<const uint8
 
     switch (datatype) {
         case Type::INT32:
-            return TypedBufferI32{value_bytes};
+            return TypedBufferI32{value_bytes, num_elements};
         case Type::INT64:
-            return TypedBufferI64{value_bytes};
+            return TypedBufferI64{value_bytes, num_elements};
         case Type::FLOAT:
-            return TypedBufferFloat{value_bytes};
+            return TypedBufferFloat{value_bytes, num_elements};
         case Type::DOUBLE:
-            return TypedBufferDouble{value_bytes};
+            return TypedBufferDouble{value_bytes, num_elements};
         case Type::INT96:
-            return TypedBufferInt96{value_bytes};
+            return TypedBufferInt96{value_bytes, num_elements};
         case Type::FIXED_LEN_BYTE_ARRAY: {
             if (!datatype_length.has_value() || datatype_length.value() <= 0) {
                 throw InvalidInputException("FIXED_LEN_BYTE_ARRAY requires a positive datatype_length");
             }
             return TypedBufferRawBytesFixedSized{
-                value_bytes, 0, RawBytesFixedSizedCodec{static_cast<size_t>(datatype_length.value())}};
+                value_bytes, num_elements, 0, RawBytesFixedSizedCodec{static_cast<size_t>(datatype_length.value())}};
         }
         case Type::BYTE_ARRAY:
-            return TypedBufferRawBytesVariableSized{value_bytes};
+            return TypedBufferRawBytesVariableSized{value_bytes, num_elements};
         default:
             throw InvalidInputException(
                 "Invalid datatype: " + std::string(to_string(datatype)));

src/processing/parquet_utils.h

@@ -113,6 +113,7 @@ std::vector<uint8_t> CompressAndJoin(
  */
 dbps::processing::TypedValuesBuffer ReinterpretValueBytesAsTypedValuesBuffer(
     tcb::span<const uint8_t> value_bytes,
+    size_t num_elements,
     Type::type datatype,
     const std::optional<int>& datatype_length,
     Encoding::type encoding);

src/processing/parquet_utils_test.cpp

@@ -657,7 +657,7 @@ TEST(ParquetUtils, Reinterpret_INT32) {
                                reinterpret_cast<const uint8_t*>(values.data()) + values.size() * sizeof(int32_t));
 
     TypedValuesBuffer result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::INT32, std::nullopt, Encoding::PLAIN);
+        bytes, values.size(), Type::INT32, std::nullopt, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferI32>(&result);
     ASSERT_NE(nullptr, buf);
@@ -676,7 +676,7 @@ TEST(ParquetUtils, Reinterpret_DOUBLE) {
                                reinterpret_cast<const uint8_t*>(values.data()) + values.size() * sizeof(double));
 
     TypedValuesBuffer result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::DOUBLE, std::nullopt, Encoding::PLAIN);
+        bytes, values.size(), Type::DOUBLE, std::nullopt, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferDouble>(&result);
     ASSERT_NE(nullptr, buf);
@@ -704,7 +704,7 @@ TEST(ParquetUtils, Reinterpret_INT96) {
     }
 
     TypedValuesBuffer result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::INT96, std::nullopt, Encoding::PLAIN);
+        bytes, expected.size(), Type::INT96, std::nullopt, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferInt96>(&result);
     ASSERT_NE(nullptr, buf);
@@ -736,7 +736,7 @@ TEST(ParquetUtils, Reinterpret_BYTE_ARRAY) {
     }
 
     TypedValuesBuffer result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::BYTE_ARRAY, std::nullopt, Encoding::PLAIN);
+        bytes, expected.size(), Type::BYTE_ARRAY, std::nullopt, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferRawBytesVariableSized>(&result);
     ASSERT_NE(nullptr, buf);
@@ -764,7 +764,7 @@ TEST(ParquetUtils, Reinterpret_FIXED_LEN_BYTE_ARRAY) {
     }
 
     TypedValuesBuffer result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::FIXED_LEN_BYTE_ARRAY, element_len, Encoding::PLAIN);
+        bytes, expected.size(), Type::FIXED_LEN_BYTE_ARRAY, element_len, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferRawBytesFixedSized>(&result);
     ASSERT_NE(nullptr, buf);
@@ -780,28 +780,28 @@ TEST(ParquetUtils, Reinterpret_FIXED_LEN_BYTE_ARRAY) {
 TEST(ParquetUtils, Reinterpret_UnsupportedEncoding) {
     std::vector<uint8_t> bytes = {0x01, 0x02, 0x03, 0x04};
     EXPECT_THROW(
-        ReinterpretValueBytesAsTypedValuesBuffer(bytes, Type::INT32, std::nullopt, Encoding::RLE),
+        ReinterpretValueBytesAsTypedValuesBuffer(bytes, 1u, Type::INT32, std::nullopt, Encoding::RLE),
         DBPSUnsupportedException);
 }
 
 TEST(ParquetUtils, Reinterpret_RLE_DICTIONARY_Throws) {
     std::vector<uint8_t> bytes = {0x01, 0x02, 0x03, 0x04};
     EXPECT_THROW(
-        ReinterpretValueBytesAsTypedValuesBuffer(bytes, Type::INT32, std::nullopt, Encoding::RLE_DICTIONARY),
+        ReinterpretValueBytesAsTypedValuesBuffer(bytes, 1u, Type::INT32, std::nullopt, Encoding::RLE_DICTIONARY),
         DBPSUnsupportedException);
 }
 
 TEST(ParquetUtils, Reinterpret_BOOLEAN_Throws) {
     std::vector<uint8_t> bytes = {0xB4};
     EXPECT_THROW(
-        ReinterpretValueBytesAsTypedValuesBuffer(bytes, Type::BOOLEAN, std::nullopt, Encoding::PLAIN),
+        ReinterpretValueBytesAsTypedValuesBuffer(bytes, 1u, Type::BOOLEAN, std::nullopt, Encoding::PLAIN),
         DBPSUnsupportedException);
 }
 
 TEST(ParquetUtils, Reinterpret_InvalidDataSize) {
     std::vector<uint8_t> bytes = {0xAA, 0xBB, 0xCC};
     auto result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::INT32, std::nullopt, Encoding::PLAIN);
+        bytes, 1u, Type::INT32, std::nullopt, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferI32>(&result);
     ASSERT_NE(nullptr, buf);
@@ -813,28 +813,29 @@ TEST(ParquetUtils, Reinterpret_InvalidDataSize) {
 TEST(ParquetUtils, Reinterpret_FIXED_LEN_BYTE_ARRAY_MissingLength_Throws) {
     std::vector<uint8_t> bytes = {0x01, 0x02, 0x03};
     EXPECT_THROW(
-        ReinterpretValueBytesAsTypedValuesBuffer(bytes, Type::FIXED_LEN_BYTE_ARRAY, std::nullopt, Encoding::PLAIN),
+        ReinterpretValueBytesAsTypedValuesBuffer(
+            bytes, 1u, Type::FIXED_LEN_BYTE_ARRAY, std::nullopt, Encoding::PLAIN),
         InvalidInputException);
 }
 
 TEST(ParquetUtils, Reinterpret_FIXED_LEN_BYTE_ARRAY_ZeroLength_Throws) {
     std::vector<uint8_t> bytes = {0x01, 0x02, 0x03};
     EXPECT_THROW(
-        ReinterpretValueBytesAsTypedValuesBuffer(bytes, Type::FIXED_LEN_BYTE_ARRAY, 0, Encoding::PLAIN),
+        ReinterpretValueBytesAsTypedValuesBuffer(bytes, 1u, Type::FIXED_LEN_BYTE_ARRAY, 0, Encoding::PLAIN),
         InvalidInputException);
 }
 
 TEST(ParquetUtils, Reinterpret_FIXED_LEN_BYTE_ARRAY_NegativeLength_Throws) {
     std::vector<uint8_t> bytes = {0x01, 0x02, 0x03};
     EXPECT_THROW(
-        ReinterpretValueBytesAsTypedValuesBuffer(bytes, Type::FIXED_LEN_BYTE_ARRAY, -1, Encoding::PLAIN),
+        ReinterpretValueBytesAsTypedValuesBuffer(bytes, 1u, Type::FIXED_LEN_BYTE_ARRAY, -1, Encoding::PLAIN),
         InvalidInputException);
 }
 
 TEST(ParquetUtils, Reinterpret_EmptyBytes_FixedSize) {
     std::vector<uint8_t> bytes;
     TypedValuesBuffer result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::DOUBLE, std::nullopt, Encoding::PLAIN);
+        bytes, 0u, Type::DOUBLE, std::nullopt, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferDouble>(&result);
     ASSERT_NE(nullptr, buf);
@@ -846,7 +847,7 @@ TEST(ParquetUtils, Reinterpret_EmptyBytes_FixedSize) {
 TEST(ParquetUtils, Reinterpret_EmptyBytes_VariableSize) {
     std::vector<uint8_t> bytes;
     TypedValuesBuffer result = ReinterpretValueBytesAsTypedValuesBuffer(
-        bytes, Type::BYTE_ARRAY, std::nullopt, Encoding::PLAIN);
+        bytes, 0u, Type::BYTE_ARRAY, std::nullopt, Encoding::PLAIN);
 
     auto* buf = std::get_if<TypedBufferRawBytesVariableSized>(&result);
     ASSERT_NE(nullptr, buf);
@@ -866,7 +867,7 @@ TEST(ParquetUtils, RoundTrip_INT32) {
         reinterpret_cast<const uint8_t*>(values.data()) + values.size() * sizeof(int32_t));
 
     auto read_buf = ReinterpretValueBytesAsTypedValuesBuffer(
-        input_bytes, Type::INT32, std::nullopt, Encoding::PLAIN);
+        input_bytes, values.size(), Type::INT32, std::nullopt, Encoding::PLAIN);
 
     auto* src = std::get_if<TypedBufferI32>(&read_buf);
     ASSERT_NE(nullptr, src);
@@ -900,7 +901,7 @@ TEST(ParquetUtils, RoundTrip_BYTE_ARRAY) {
     }
 
     auto read_buf = ReinterpretValueBytesAsTypedValuesBuffer(
-        input_bytes, Type::BYTE_ARRAY, std::nullopt, Encoding::PLAIN);
+        input_bytes, payloads.size(), Type::BYTE_ARRAY, std::nullopt, Encoding::PLAIN);
 
     auto* src = std::get_if<TypedBufferRawBytesVariableSized>(&read_buf);
     ASSERT_NE(nullptr, src);
@@ -931,7 +932,7 @@ TEST(ParquetUtils, RoundTrip_FIXED_LEN_BYTE_ARRAY) {
     }
 
     auto read_buf = ReinterpretValueBytesAsTypedValuesBuffer(
-        input_bytes, Type::FIXED_LEN_BYTE_ARRAY, element_len, Encoding::PLAIN);
+        input_bytes, payloads.size(), Type::FIXED_LEN_BYTE_ARRAY, element_len, Encoding::PLAIN);
 
     auto* src = std::get_if<TypedBufferRawBytesFixedSized>(&read_buf);
     ASSERT_NE(nullptr, src);