commit 9eecaaf

Author: Omega Gamage <[email protected]>
Date:   Tue Feb 18 14:23:08 2020 +0530

    used std::map instead of std::unordered_map to store num_data_pages

commit 6822fc6
Author: Omega Gamage <[email protected]>
Date:   Mon Feb 17 11:45:03 2020 +0530

    remove default arguments for page number counts in ColumnChunkMetaDataBuilder::Finish

commit 49e1861
Author: Omega Gamage <[email protected]>
Date:   Fri Feb 14 16:22:41 2020 +0530

    Added the class PageEncodingStats to types.h

commit 9dc5b6b
Merge: 9b776f3 d65a71a
Author: Omega Gamage <[email protected]>
Date:   Wed Feb 12 12:25:50 2020 +0530

    resolved merge conflicts

commit 9b776f3
Author: Omega Gamage <[email protected]>
Date:   Thu Feb 6 16:12:31 2020 +0530

    PARQUET-1780: [C++] Set ColumnMetadata.encoding_stats field

    Fixed lint errors

    Use std::map to store datapage count

    Added unit test to test encoding_stats

commit d65a71a
Author: Omega Gamage <[email protected]>
Date:   Thu Feb 6 20:11:09 2020 +0530

    Fixed lint errors

commit 053ce4d
Author: Omega Gamage <[email protected]>
Date:   Thu Feb 6 16:12:31 2020 +0530

    PARQUET-1780: [C++] Set ColumnMetadata.encoding_stats field

Author: omega-bigstream

cpp/src/parquet/column_writer.cc

@@ -20,6 +20,7 @@
 #include <algorithm>
 #include <cstdint>
 #include <cstring>
+#include <map>
 #include <memory>
 #include <string>
 #include <utility>
@@ -167,7 +168,8 @@ class SerializedPageWriter : public PageWriter {
         column_ordinal_(column_chunk_ordinal),
         meta_encryptor_(std::move(meta_encryptor)),
         data_encryptor_(std::move(data_encryptor)),
-        encryption_buffer_(AllocateBuffer(pool, 0)) {
+        encryption_buffer_(AllocateBuffer(pool, 0)),
+        num_dict_pages_(0) {
     if (data_encryptor_ != nullptr || meta_encryptor_ != nullptr) {
       InitEncryption();
     }
@@ -226,6 +228,7 @@ class SerializedPageWriter : public PageWriter {
 
     total_uncompressed_size_ += uncompressed_size + header_size;
     total_compressed_size_ += output_data_len + header_size;
+    num_dict_pages_++;
 
     PARQUET_ASSIGN_OR_THROW(int64_t final_pos, sink_->Tell());
     return final_pos - start_pos;
@@ -238,7 +241,7 @@ class SerializedPageWriter : public PageWriter {
     // index_page_offset = -1 since they are not supported
     metadata_->Finish(num_values_, dictionary_page_offset_, -1, data_page_offset_,
                       total_compressed_size_, total_uncompressed_size_, has_dictionary,
-                      fallback, meta_encryptor_);
+                      fallback, num_dict_pages_, num_data_pages_, meta_encryptor_);
     // Write metadata at end of column chunk
     metadata_->WriteTo(sink_.get());
   }
@@ -310,7 +313,11 @@ class SerializedPageWriter : public PageWriter {
     total_uncompressed_size_ += uncompressed_size + header_size;
     total_compressed_size_ += output_data_len + header_size;
     num_values_ += page.num_values();
-
+    if (num_data_pages_.find(page.encoding()) != num_data_pages_.end()) {
+      num_data_pages_[page.encoding()]++;
+    } else {
+      num_data_pages_[page.encoding()] = 1;
+    }
     ++page_ordinal_;
     PARQUET_ASSIGN_OR_THROW(int64_t current_pos, sink_->Tell());
     return current_pos - start_pos;
@@ -405,6 +412,9 @@ class SerializedPageWriter : public PageWriter {
   std::shared_ptr<Encryptor> data_encryptor_;
 
   std::shared_ptr<ResizableBuffer> encryption_buffer_;
+
+  int32_t num_dict_pages_;
+  std::map<Encoding::type, int32_t> num_data_pages_;
 };
 
 // This implementation of the PageWriter writes to the final sink on Close .
@@ -441,7 +451,8 @@ class BufferedPageWriter : public PageWriter {
     metadata_->Finish(pager_->num_values(), dictionary_page_offset, -1,
                       pager_->data_page_offset() + final_position,
                       pager_->total_compressed_size(), pager_->total_uncompressed_size(),
-                      has_dictionary, fallback, pager_->meta_encryptor_);
+                      has_dictionary, fallback, pager_->num_dict_pages_,
+                      pager_->num_data_pages_, pager_->meta_encryptor_);
 
     // Write metadata at end of column chunk
     metadata_->WriteTo(in_memory_sink_.get());

cpp/src/parquet/column_writer_test.cc

@@ -185,6 +185,39 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
           {Encoding::RLE_DICTIONARY, Encoding::PLAIN, Encoding::RLE, Encoding::PLAIN});
       ASSERT_EQ(encodings, expected);
     }
+
+    std::vector<parquet::PageEncodingStats> encoding_stats =
+        this->metadata_encoding_stats();
+    if (this->type_num() == Type::BOOLEAN) {
+      ASSERT_EQ(encoding_stats[0].encoding, Encoding::PLAIN);
+      ASSERT_EQ(encoding_stats[0].page_type, PageType::DATA_PAGE);
+      ASSERT_EQ(encoding_stats[1].encoding, Encoding::RLE);
+      ASSERT_EQ(encoding_stats[1].page_type, PageType::DATA_PAGE);
+    } else if (version == ParquetVersion::PARQUET_1_0) {
+      std::vector<Encoding::type> expected(
+          {Encoding::PLAIN_DICTIONARY, Encoding::PLAIN, Encoding::RLE, Encoding::PLAIN});
+      for (size_t i = 0; i < encoding_stats.size(); i++) {
+        if (i == 0) {
+          ASSERT_EQ(encoding_stats[i].encoding, expected[i]);
+          ASSERT_EQ(encoding_stats[i].page_type, PageType::DICTIONARY_PAGE);
+        } else {
+          ASSERT_EQ(encoding_stats[i].encoding, expected[i]);
+          ASSERT_EQ(encoding_stats[i].page_type, PageType::DATA_PAGE);
+        }
+      }
+    } else {
+      std::vector<Encoding::type> expected(
+          {Encoding::RLE_DICTIONARY, Encoding::PLAIN, Encoding::RLE, Encoding::PLAIN});
+      for (size_t i = 0; i < encoding_stats.size(); i++) {
+        if (i == 0) {
+          ASSERT_EQ(encoding_stats[i].encoding, expected[i]);
+          ASSERT_EQ(encoding_stats[i].page_type, PageType::DICTIONARY_PAGE);
+        } else {
+          ASSERT_EQ(encoding_stats[i].encoding, expected[i]);
+          ASSERT_EQ(encoding_stats[i].page_type, PageType::DATA_PAGE);
+        }
+      }
+    }
   }
 
   void WriteRequiredWithSettings(Encoding::type encoding, Compression::type compression,
@@ -273,6 +306,15 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
     return metadata_accessor->encodings();
   }
 
+  std::vector<parquet::PageEncodingStats> metadata_encoding_stats() {
+    // Metadata accessor must be created lazily.
+    // This is because the ColumnChunkMetaData semantics dictate the metadata object is
+    // complete (no changes to the metadata buffer can be made after instantiation)
+    auto metadata_accessor =
+        ColumnChunkMetaData::Make(metadata_->contents(), this->descr_);
+    return metadata_accessor->encoding_stats();
+  }
+
  protected:
   int64_t values_read_;
   // Keep the reader alive as for ByteArray the lifetime of the ByteArray

cpp/src/parquet/metadata.cc

@@ -214,6 +214,14 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
     for (auto encoding : column_metadata_->encodings) {
       encodings_.push_back(FromThrift(encoding));
     }
+    auto fromthrift = [](format::PageEncodingStats page_encoding_stats) {
+      return parquet::PageEncodingStats(FromThrift(page_encoding_stats.page_type),
+                                        FromThrift(page_encoding_stats.encoding),
+                                        page_encoding_stats.count);
+    };
+    for (auto encoding_stats : column_metadata_->encoding_stats) {
+      encoding_stats_.push_back(fromthrift(encoding_stats));
+    }
     possible_stats_ = nullptr;
   }
   // column chunk
@@ -257,6 +265,8 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
 
   const std::vector<Encoding::type>& encodings() const { return encodings_; }
 
+  const std::vector<PageEncodingStats>& encoding_stats() const { return encoding_stats_; }
+
   inline bool has_dictionary_page() const {
     return column_metadata_->__isset.dictionary_page_offset;
   }
@@ -293,6 +303,7 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
  private:
   mutable std::shared_ptr<Statistics> possible_stats_;
   std::vector<Encoding::type> encodings_;
+  std::vector<PageEncodingStats> encoding_stats_;
   const format::ColumnChunk* column_;
   const format::ColumnMetaData* column_metadata_;
   format::ColumnMetaData decrypted_metadata_;
@@ -367,6 +378,10 @@ const std::vector<Encoding::type>& ColumnChunkMetaData::encodings() const {
   return impl_->encodings();
 }
 
+const std::vector<PageEncodingStats>& ColumnChunkMetaData::encoding_stats() const {
+  return impl_->encoding_stats();
+}
+
 int64_t ColumnChunkMetaData::total_uncompressed_size() const {
   return impl_->total_uncompressed_size();
 }
@@ -966,7 +981,9 @@ class ColumnChunkMetaDataBuilder::ColumnChunkMetaDataBuilderImpl {
   void Finish(int64_t num_values, int64_t dictionary_page_offset,
               int64_t index_page_offset, int64_t data_page_offset,
               int64_t compressed_size, int64_t uncompressed_size, bool has_dictionary,
-              bool dictionary_fallback, const std::shared_ptr<Encryptor>& encryptor) {
+              bool dictionary_fallback, int32_t num_dict_pages,
+              std::map<Encoding::type, int32_t>& num_data_pages,
+              const std::shared_ptr<Encryptor>& encryptor) {
     if (dictionary_page_offset > 0) {
       column_chunk_->meta_data.__set_dictionary_page_offset(dictionary_page_offset);
       column_chunk_->__set_file_offset(dictionary_page_offset + compressed_size);
@@ -983,23 +1000,62 @@ class ColumnChunkMetaDataBuilder::ColumnChunkMetaDataBuilderImpl {
     column_chunk_->meta_data.__set_total_compressed_size(compressed_size);
 
     std::vector<format::Encoding::type> thrift_encodings;
+    std::vector<format::PageEncodingStats> thrift_encoding_stats;
     if (has_dictionary) {
       thrift_encodings.push_back(ToThrift(properties_->dictionary_index_encoding()));
+      format::PageEncodingStats dict_page_stats;
+      dict_page_stats.__set_page_type(format::PageType::DICTIONARY_PAGE);
+      dict_page_stats.__set_encoding(ToThrift(properties_->dictionary_index_encoding()));
+      dict_page_stats.__set_count(num_dict_pages);
+      thrift_encoding_stats.push_back(dict_page_stats);
+      // Add DataPage stats
+      format::PageEncodingStats data_page_stats;
       if (properties_->version() == ParquetVersion::PARQUET_1_0) {
         thrift_encodings.push_back(ToThrift(Encoding::PLAIN));
+        data_page_stats.__set_page_type(format::PageType::DATA_PAGE);
+        data_page_stats.__set_encoding(ToThrift(Encoding::PLAIN));
+        data_page_stats.__set_count(num_data_pages[Encoding::PLAIN]);
       } else {
         thrift_encodings.push_back(ToThrift(properties_->dictionary_page_encoding()));
+        data_page_stats.__set_page_type(format::PageType::DATA_PAGE);
+        data_page_stats.__set_encoding(ToThrift(properties_->dictionary_page_encoding()));
+        data_page_stats.__set_count(
+            num_data_pages[properties_->dictionary_page_encoding()]);
       }
+      thrift_encoding_stats.push_back(data_page_stats);
     } else {  // Dictionary not enabled
       thrift_encodings.push_back(ToThrift(properties_->encoding(column_->path())));
+      // Add DataPage stats
+      format::PageEncodingStats data_page_stats;
+      data_page_stats.__set_page_type(format::PageType::DATA_PAGE);
+      if (column_->physical_type() == Type::BOOLEAN) {
+        data_page_stats.__set_encoding(ToThrift(Encoding::PLAIN));
+        data_page_stats.__set_count(num_data_pages[Encoding::PLAIN]);
+      } else {
+        data_page_stats.__set_encoding(ToThrift(properties_->encoding(column_->path())));
+        data_page_stats.__set_count(
+            num_data_pages[properties_->encoding(column_->path())]);
+      }
+      thrift_encoding_stats.push_back(data_page_stats);
     }
     thrift_encodings.push_back(ToThrift(Encoding::RLE));
+    format::PageEncodingStats data_page_stats;
+    data_page_stats.__set_page_type(format::PageType::DATA_PAGE);
+    data_page_stats.__set_encoding(ToThrift(Encoding::RLE));
+    data_page_stats.__set_count(num_data_pages[Encoding::RLE]);
+    thrift_encoding_stats.push_back(data_page_stats);
     // Only PLAIN encoding is supported for fallback in V1
     // TODO(majetideepak): Use user specified encoding for V2
     if (dictionary_fallback) {
       thrift_encodings.push_back(ToThrift(Encoding::PLAIN));
+      format::PageEncodingStats fallback_page_stats;
+      fallback_page_stats.__set_page_type(format::PageType::DATA_PAGE);
+      fallback_page_stats.__set_encoding(ToThrift(Encoding::PLAIN));
+      fallback_page_stats.__set_count(num_data_pages[Encoding::PLAIN]);
+      thrift_encoding_stats.push_back(fallback_page_stats);
     }
     column_chunk_->meta_data.__set_encodings(thrift_encodings);
+    column_chunk_->meta_data.__set_encoding_stats(thrift_encoding_stats);
 
     const auto& encrypt_md =
         properties_->column_encryption_properties(column_->path()->ToDotString());
@@ -1122,11 +1178,12 @@ void ColumnChunkMetaDataBuilder::Finish(int64_t num_values,
                                         int64_t index_page_offset,
                                         int64_t data_page_offset, int64_t compressed_size,
                                         int64_t uncompressed_size, bool has_dictionary,
-                                        bool dictionary_fallback,
+                                        bool dictionary_fallback, int32_t num_dict_pages,
+                                        std::map<Encoding::type, int32_t> num_data_pages,
                                         const std::shared_ptr<Encryptor>& encryptor) {
   impl_->Finish(num_values, dictionary_page_offset, index_page_offset, data_page_offset,
                 compressed_size, uncompressed_size, has_dictionary, dictionary_fallback,
-                encryptor);
+                num_dict_pages, num_data_pages, encryptor);
 }
 
 void ColumnChunkMetaDataBuilder::WriteTo(::arrow::io::OutputStream* sink) {

cpp/src/parquet/metadata.h

@@ -21,6 +21,7 @@
 #include <map>
 #include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "arrow/util/key_value_metadata.h"
@@ -150,6 +151,7 @@ class PARQUET_EXPORT ColumnChunkMetaData {
   bool can_decompress() const;
 
   const std::vector<Encoding::type>& encodings() const;
+  const std::vector<PageEncodingStats>& encoding_stats() const;
   bool has_dictionary_page() const;
   int64_t dictionary_page_offset() const;
   int64_t data_page_offset() const;
@@ -319,7 +321,8 @@ class PARQUET_EXPORT ColumnChunkMetaDataBuilder {
   void Finish(int64_t num_values, int64_t dictionary_page_offset,
               int64_t index_page_offset, int64_t data_page_offset,
               int64_t compressed_size, int64_t uncompressed_size, bool has_dictionary,
-              bool dictionary_fallback,
+              bool dictionary_fallback, int32_t num_dict_pages,
+              std::map<Encoding::type, int32_t>,
               const std::shared_ptr<Encryptor>& encryptor = NULLPTR);
 
   // The metadata contents, suitable for passing to ColumnChunkMetaData::Make

cpp/src/parquet/metadata_test.cc

@@ -35,18 +35,19 @@ std::unique_ptr<parquet::FileMetaData> GenerateTableMetaData(
     EncodedStatistics stats_int, EncodedStatistics stats_float) {
   auto f_builder = FileMetaDataBuilder::Make(&schema, props);
   auto rg1_builder = f_builder->AppendRowGroup();
-
   // Write the metadata
   // rowgroup1 metadata
   auto col1_builder = rg1_builder->NextColumnChunk();
   auto col2_builder = rg1_builder->NextColumnChunk();
   // column metadata
+  std::map<Encoding::type, int32_t> encoding_stats(
+      {{Encoding::RLE_DICTIONARY, 1}, {Encoding::PLAIN, 1}, {Encoding::RLE, 1}});
   stats_int.set_is_signed(true);
   col1_builder->SetStatistics(stats_int);
   stats_float.set_is_signed(true);
   col2_builder->SetStatistics(stats_float);
-  col1_builder->Finish(nrows / 2, 4, 0, 10, 512, 600, true, false);
-  col2_builder->Finish(nrows / 2, 24, 0, 30, 512, 600, true, false);
+  col1_builder->Finish(nrows / 2, 4, 0, 10, 512, 600, true, false, 1, encoding_stats);
+  col2_builder->Finish(nrows / 2, 24, 0, 30, 512, 600, true, false, 1, encoding_stats);
 
   rg1_builder->set_num_rows(nrows / 2);
   rg1_builder->Finish(1024);
@@ -58,8 +59,8 @@ std::unique_ptr<parquet::FileMetaData> GenerateTableMetaData(
   // column metadata
   col1_builder->SetStatistics(stats_int);
   col2_builder->SetStatistics(stats_float);
-  col1_builder->Finish(nrows / 2, 6, 0, 10, 512, 600, true, false);
-  col2_builder->Finish(nrows / 2, 16, 0, 26, 512, 600, true, false);
+  col1_builder->Finish(nrows / 2, 6, 0, 10, 512, 600, true, false, 1, encoding_stats);
+  col2_builder->Finish(nrows / 2, 16, 0, 26, 512, 600, true, false, 1, encoding_stats);
 
   rg2_builder->set_num_rows(nrows / 2);
   rg2_builder->Finish(1024);
@@ -155,6 +156,8 @@ TEST(Metadata, TestBuildAccess) {
     ASSERT_EQ(24, rg1_column2->dictionary_page_offset());
     ASSERT_EQ(10, rg1_column1->data_page_offset());
     ASSERT_EQ(30, rg1_column2->data_page_offset());
+    ASSERT_EQ(3, rg1_column1->encoding_stats().size());
+    ASSERT_EQ(3, rg1_column2->encoding_stats().size());
 
     auto rg2_accessor = f_accessors[loop_index]->RowGroup(1);
     ASSERT_EQ(2, rg2_accessor->num_columns());
@@ -187,6 +190,8 @@ TEST(Metadata, TestBuildAccess) {
     ASSERT_EQ(16, rg2_column2->dictionary_page_offset());
     ASSERT_EQ(10, rg2_column1->data_page_offset());
     ASSERT_EQ(26, rg2_column2->data_page_offset());
+    ASSERT_EQ(3, rg2_column1->encoding_stats().size());
+    ASSERT_EQ(3, rg2_column2->encoding_stats().size());
 
     // Test FileMetaData::set_file_path
     ASSERT_TRUE(rg2_column1->file_path().empty());

cpp/src/parquet/thrift_internal.h

@@ -82,6 +82,10 @@ static inline Encoding::type FromThrift(format::Encoding::type type) {
   return static_cast<Encoding::type>(type);
 }
 
+static inline PageType::type FromThrift(format::PageType::type type) {
+  return static_cast<PageType::type>(type);
+}
+
 static inline AadMetadata FromThrift(format::AesGcmV1 aesGcmV1) {
   return AadMetadata{aesGcmV1.aad_prefix, aesGcmV1.aad_file_unique,
                      aesGcmV1.supply_aad_prefix};

cpp/src/parquet/types.cc

@@ -335,6 +335,24 @@ SortOrder::type GetSortOrder(const std::shared_ptr<const LogicalType>& logical_t
 ColumnOrder ColumnOrder::undefined_ = ColumnOrder(ColumnOrder::UNDEFINED);
 ColumnOrder ColumnOrder::type_defined_ = ColumnOrder(ColumnOrder::TYPE_DEFINED_ORDER);
 
+void PageEncodingStats::set_page_type(const PageType::type val) { this->page_type = val; }
+
+void PageEncodingStats::set_encoding(const Encoding::type val) { this->encoding = val; }
+
+void PageEncodingStats::set_count(const int32_t val) { this->count = val; }
+
+PageEncodingStats::PageEncodingStats(const PageEncodingStats& obj) {
+  page_type = obj.page_type;
+  encoding = obj.encoding;
+  count = obj.count;
+}
+PageEncodingStats& PageEncodingStats::operator=(const PageEncodingStats& obj) {
+  page_type = obj.page_type;
+  encoding = obj.encoding;
+  count = obj.count;
+  return *this;
+}
+
 // Static methods for LogicalType class
 
 std::shared_ptr<const LogicalType> LogicalType::FromConvertedType(