GH-33115: [C++] Parquet Implement crc in reading and writing Page for DATA_PAGE (v1)

cpp/src/arrow/CMakeLists.txt

@@ -206,6 +206,7 @@ set(ARROW_SRCS
     util/compression.cc
     util/counting_semaphore.cc
     util/cpu_info.cc
+    util/crc32.cc
     util/debug.cc
     util/decimal.cc
     util/delimiting.cc

cpp/src/arrow/util/CMakeLists.txt

@@ -90,6 +90,8 @@ add_arrow_test(threading-utility-test
                test_common.cc
                thread_pool_test.cc)
 
+add_arrow_test(crc32-test SOURCES crc32_test.cc)
+
 add_arrow_benchmark(bit_block_counter_benchmark)
 add_arrow_benchmark(bit_util_benchmark)
 add_arrow_benchmark(bitmap_reader_benchmark)

cpp/src/arrow/util/crc32.h

@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <cstddef>
+#include <cstdint>
+
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+namespace internal {
+
+/// \brief Compute the CRC32 checksum of the given data
+///
+/// This function computes CRC32 with the polynomial 0x04C11DB7,
+/// as used in zlib and others (note this is different from CRC32C).
+/// To compute a running CRC32, pass the previous value in `prev`,
+/// otherwise `prev` should be 0.
+ARROW_EXPORT
+uint32_t crc32(uint32_t prev, const void* data, size_t length);
+
+}  // namespace internal
+}  // namespace arrow

cpp/src/arrow/util/crc32_test.cc

@@ -0,0 +1,110 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <array>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <random>
+
+#include <gtest/gtest.h>
+#include <boost/crc.hpp>
+
+#include "arrow/util/crc32.h"
+
+namespace arrow {
+
+TEST(Crc32Test, Basic) {
+  // use the string "123456789" in ASCII as test data.
+  constexpr uint32_t TEST_CRC32_RESULT = 0xCBF43926;
+  constexpr size_t TEST_CRC32_LENGTH = 9;
+  std::array<unsigned char, TEST_CRC32_LENGTH> std_data = {0x31, 0x32, 0x33, 0x34, 0x35,
+                                                           0x36, 0x37, 0x38, 0x39};
+  size_t const std_data_len = sizeof(std_data) / sizeof(std_data[0]);
+  EXPECT_EQ(TEST_CRC32_RESULT, internal::crc32(0, &std_data[0], std_data_len));
+
+  for (size_t i = 1; i < std_data_len - 1; ++i) {
+    uint32_t crc1 = internal::crc32(0, &std_data[0], i);
+    EXPECT_EQ(TEST_CRC32_RESULT, internal::crc32(crc1, &std_data[i], std_data_len - i));
+  }
+}
+
+TEST(Crc32Test, matchesBoost32Type) {
+  const size_t BUFFER_SIZE = 512 * 1024 * sizeof(uint64_t);
+  std::vector<uint8_t> buffer;
+  buffer.resize(BUFFER_SIZE, 0);
+
+  // Populate a buffer with a deterministic pattern
+  // on which to compute checksums
+  std::random_device r;
+  std::seed_seq seed{r(), r(), r(), r(), r(), r(), r(), r()};
+  std::mt19937 gen(seed);
+  // N4659 29.6.1.1 [rand.req.genl]/1e requires one of short, int, long, long long,
+  // unsigned short, unsigned int, unsigned long, or unsigned long long
+  std::uniform_int_distribution<uint32_t> dist;
+
+  for (size_t i = 0; i < BUFFER_SIZE; ++i) {
+    buffer[i] = static_cast<uint8_t>(dist(gen));
+  }
+
+  struct TestCrcGroup {
+    size_t offset;
+    size_t length;
+  };
+
+  // NOLINTNEXTLINE
+  TestCrcGroup testCrcGroups[] = {
+      // Zero-byte input
+      {0, 0},
+      {8, 1},
+      {8, 2},
+      {8, 3},
+      {8, 4},
+      {8, 5},
+      {8, 6},
+      {8, 7},
+      {9, 1},
+      {10, 2},
+      {11, 3},
+      {12, 4},
+      {13, 5},
+      {14, 6},
+      {15, 7},
+      {8, 8},
+      {8, 9},
+      {8, 10},
+      {8, 11},
+      {8, 12},
+      {8, 13},
+      {8, 14},
+      {8, 15},
+      {8, 16},
+      {8, 17},
+      // Much larger inputs
+      {0, BUFFER_SIZE},
+      {1, BUFFER_SIZE / 2},
+  };
+
+  for (TestCrcGroup group : testCrcGroups) {
+    uint32_t crc = internal::crc32(0, &buffer[group.offset], group.length);
+    boost::crc_32_type boost_crc;
+    boost_crc.process_bytes(&buffer[group.offset], group.length);
+    EXPECT_EQ(boost_crc.checksum(), crc);
+  }
+}
+
+}  // namespace arrow

cpp/src/parquet/arrow/path_internal.cc

@@ -201,7 +201,7 @@ struct PathWriteContext {
 
   // Incorporates |range| into visited elements. If the |range| is contiguous
   // with the last range, extend the last range, otherwise add |range| separately
-  // tot he list.
+  // to the list.
   void RecordPostListVisit(const ElementRange& range) {
     if (!visited_elements.empty() && range.start == visited_elements.back().end) {
       visited_elements.back().end = range.end;

cpp/src/parquet/column_reader.cc

@@ -39,6 +39,7 @@
 #include "arrow/util/bit_util.h"
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/compression.h"
+#include "arrow/util/crc32.h"
 #include "arrow/util/int_util_overflow.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/rle_encoding.h"
@@ -471,6 +472,20 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
       ParquetException::EofException(ss.str());
     }
 
+    const PageType::type page_type = LoadEnumSafe(&current_page_header_.type);
+
+    // TODO(PARQUET-594) crc checksum for DATA_PAGE_V2 and DICT_PAGE
+    if (properties_.page_checksum_verification() && page_type == PageType::DATA_PAGE &&
+        current_page_header_.__isset.crc) {
+      // verify crc
+      uint32_t checksum =
+          ::arrow::internal::crc32(/* prev */ 0, page_buffer->data(), compressed_len);
+      if (static_cast<int32_t>(checksum) != current_page_header_.crc) {
+        throw ParquetException(
+            "could not verify page integrity, CRC checksum verification failed");
+      }
+    }
+
     // Decrypt it if we need to
     if (crypto_ctx_.data_decryptor != nullptr) {
       PARQUET_THROW_NOT_OK(decryption_buffer_->Resize(
@@ -482,8 +497,6 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
       page_buffer = decryption_buffer_;
     }
 
-    // Uncompress and construct the pages to return.
-    const PageType::type page_type = LoadEnumSafe(&current_page_header_.type);
     if (page_type == PageType::DICTIONARY_PAGE) {
       crypto_ctx_.start_decrypt_with_dictionary_page = false;
       const format::DictionaryPageHeader& dict_header =

cpp/src/parquet/column_writer.cc

@@ -37,6 +37,7 @@
 #include "arrow/util/bitmap_ops.h"
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/compression.h"
+#include "arrow/util/crc32.h"
 #include "arrow/util/endian.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/rle_encoding.h"
@@ -248,6 +249,7 @@ class SerializedPageWriter : public PageWriter {
   SerializedPageWriter(std::shared_ptr<ArrowOutputStream> sink, Compression::type codec,
                        int compression_level, ColumnChunkMetaDataBuilder* metadata,
                        int16_t row_group_ordinal, int16_t column_chunk_ordinal,
+                       bool use_page_checksum_verification,
                        MemoryPool* pool = ::arrow::default_memory_pool(),
                        std::shared_ptr<Encryptor> meta_encryptor = nullptr,
                        std::shared_ptr<Encryptor> data_encryptor = nullptr)
@@ -262,6 +264,7 @@ class SerializedPageWriter : public PageWriter {
         page_ordinal_(0),
         row_group_ordinal_(row_group_ordinal),
         column_ordinal_(column_chunk_ordinal),
+        page_checksum_verification_(use_page_checksum_verification),
         meta_encryptor_(std::move(meta_encryptor)),
         data_encryptor_(std::move(data_encryptor)),
         encryption_buffer_(AllocateBuffer(pool, 0)) {
@@ -379,7 +382,13 @@ class SerializedPageWriter : public PageWriter {
     format::PageHeader page_header;
     page_header.__set_uncompressed_page_size(static_cast<int32_t>(uncompressed_size));
     page_header.__set_compressed_page_size(static_cast<int32_t>(output_data_len));
-    // TODO(PARQUET-594) crc checksum
+
+    // TODO(PARQUET-594) crc checksum for DATA_PAGE_V2 and DICT_PAGE
+    if (page_checksum_verification_ && page.type() == PageType::DATA_PAGE) {
+      uint32_t crc32 =
+          ::arrow::internal::crc32(/* prev */ 0, output_data_buffer, output_data_len);
+      page_header.__set_crc(static_cast<int32_t>(crc32));
+    }
 
     if (page.type() == PageType::DATA_PAGE) {
       const DataPageV1& v1_page = checked_cast<const DataPageV1&>(page);
@@ -425,7 +434,7 @@ class SerializedPageWriter : public PageWriter {
     page_header.__set_data_page_header(data_page_header);
   }
 
-  void SetDataPageV2Header(format::PageHeader& page_header, const DataPageV2 page) {
+  void SetDataPageV2Header(format::PageHeader& page_header, const DataPageV2& page) {
     format::DataPageHeaderV2 data_page_header;
     data_page_header.__set_num_values(page.num_values());
     data_page_header.__set_num_nulls(page.num_nulls());
@@ -456,6 +465,8 @@ class SerializedPageWriter : public PageWriter {
 
   int64_t total_uncompressed_size() { return total_uncompressed_size_; }
 
+  bool page_checksum_verification() { return page_checksum_verification_; }
+
  private:
   // To allow UpdateEncryption on Close
   friend class BufferedPageWriter;
@@ -520,6 +531,7 @@ class SerializedPageWriter : public PageWriter {
   int32_t page_ordinal_;
   int16_t row_group_ordinal_;
   int16_t column_ordinal_;
+  bool page_checksum_verification_;
 
   std::unique_ptr<ThriftSerializer> thrift_serializer_;
 
@@ -544,15 +556,16 @@ class BufferedPageWriter : public PageWriter {
   BufferedPageWriter(std::shared_ptr<ArrowOutputStream> sink, Compression::type codec,
                      int compression_level, ColumnChunkMetaDataBuilder* metadata,
                      int16_t row_group_ordinal, int16_t current_column_ordinal,
+                     bool use_page_checksum_verification,
                      MemoryPool* pool = ::arrow::default_memory_pool(),
                      std::shared_ptr<Encryptor> meta_encryptor = nullptr,
                      std::shared_ptr<Encryptor> data_encryptor = nullptr)
       : final_sink_(std::move(sink)), metadata_(metadata), has_dictionary_pages_(false) {
     in_memory_sink_ = CreateOutputStream(pool);
     pager_ = std::make_unique<SerializedPageWriter>(
         in_memory_sink_, codec, compression_level, metadata, row_group_ordinal,
-        current_column_ordinal, pool, std::move(meta_encryptor),
-        std::move(data_encryptor));
+        current_column_ordinal, use_page_checksum_verification, pool,
+        std::move(meta_encryptor), std::move(data_encryptor));
   }
 
   int64_t WriteDictionaryPage(const DictionaryPage& page) override {
@@ -606,15 +619,17 @@ std::unique_ptr<PageWriter> PageWriter::Open(
     int compression_level, ColumnChunkMetaDataBuilder* metadata,
     int16_t row_group_ordinal, int16_t column_chunk_ordinal, MemoryPool* pool,
     bool buffered_row_group, std::shared_ptr<Encryptor> meta_encryptor,
-    std::shared_ptr<Encryptor> data_encryptor) {
+    std::shared_ptr<Encryptor> data_encryptor, bool page_write_checksum_enabled) {
   if (buffered_row_group) {
-    return std::make_unique<BufferedPageWriter>(
+    return std::unique_ptr<PageWriter>(new BufferedPageWriter(
         std::move(sink), codec, compression_level, metadata, row_group_ordinal,
-        column_chunk_ordinal, pool, std::move(meta_encryptor), std::move(data_encryptor));
+        column_chunk_ordinal, page_write_checksum_enabled, pool,
+        std::move(meta_encryptor), std::move(data_encryptor)));
   } else {
-    return std::make_unique<SerializedPageWriter>(
+    return std::unique_ptr<PageWriter>(new SerializedPageWriter(
         std::move(sink), codec, compression_level, metadata, row_group_ordinal,
-        column_chunk_ordinal, pool, std::move(meta_encryptor), std::move(data_encryptor));
+        column_chunk_ordinal, page_write_checksum_enabled, pool,
+        std::move(meta_encryptor), std::move(data_encryptor)));
   }
 }
 

cpp/src/parquet/column_writer.h

@@ -90,7 +90,8 @@ class PARQUET_EXPORT PageWriter {
       ::arrow::MemoryPool* pool = ::arrow::default_memory_pool(),
       bool buffered_row_group = false,
       std::shared_ptr<Encryptor> header_encryptor = NULLPTR,
-      std::shared_ptr<Encryptor> data_encryptor = NULLPTR);
+      std::shared_ptr<Encryptor> data_encryptor = NULLPTR,
+      bool page_write_checksum_enabled = false);
 
   // The Column Writer decides if dictionary encoding is used if set and
   // if the dictionary encoding has fallen back to default encoding on reaching dictionary