Include row group level stats when writing ORC files

cpp/src/io/orc/orc.cpp

@@ -209,43 +209,55 @@ void ProtobufWriter::put_row_index_entry(int32_t present_blk,
                                          int32_t data_ofs,
                                          int32_t data2_blk,
                                          int32_t data2_ofs,
-                                         TypeKind kind)
+                                         TypeKind kind,
+                                         ColStatsBlob const* stats)
 {
   size_t sz = 0, lpos;
-  putb(1 * 8 + PB_TYPE_FIXEDLEN);  // 1:RowIndex.entry
+  put_byte(1 * 8 + PB_TYPE_FIXEDLEN);  // 1:RowIndex.entry
   lpos = m_buf->size();
-  putb(0xcd);                      // sz+2
-  putb(1 * 8 + PB_TYPE_FIXEDLEN);  // 1:positions[packed=true]
-  putb(0xcd);                      // sz
+  put_byte(0xcd);                      // sz+2
+  put_byte(1 * 8 + PB_TYPE_FIXEDLEN);  // 1:positions[packed=true]
+  put_byte(0xcd);                      // sz
   if (present_blk >= 0) sz += put_uint(present_blk);
   if (present_ofs >= 0) {
     sz += put_uint(present_ofs) + 2;
-    putb(0);  // run pos = 0
-    putb(0);  // bit pos = 0
+    put_byte(0);  // run pos = 0
+    put_byte(0);  // bit pos = 0
   }
   if (data_blk >= 0) { sz += put_uint(data_blk); }
   if (data_ofs >= 0) {
     sz += put_uint(data_ofs);
     if (kind != STRING && kind != FLOAT && kind != DOUBLE && kind != DECIMAL) {
-      putb(0);  // RLE run pos always zero (assumes RLE aligned with row index boundaries)
+      // RLE run pos always zero (assumes RLE aligned with row index boundaries)
+      put_byte(0);
       sz++;
       if (kind == BOOLEAN) {
-        putb(0);  // bit position in byte, always zero
+        // bit position in byte, always zero
+        put_byte(0);
         sz++;
       }
     }
   }
-  if (kind !=
-      INT)  // INT kind can be passed in to bypass 2nd stream index (dictionary length streams)
-  {
+  // INT kind can be passed in to bypass 2nd stream index (dictionary length streams)
+  if (kind != INT) {
     if (data2_blk >= 0) { sz += put_uint(data2_blk); }
     if (data2_ofs >= 0) {
       sz += put_uint(data2_ofs) + 1;
-      putb(0);  // RLE run pos always zero (assumes RLE aligned with row index boundaries)
+      // RLE run pos always zero (assumes RLE aligned with row index boundaries)
+      put_byte(0);
     }
   }
-  m_buf->data()[lpos]     = (uint8_t)(sz + 2);
+  // size of the field 1
   m_buf->data()[lpos + 2] = (uint8_t)(sz);
+
+  if (stats != nullptr) {
+    sz += put_uint(2 * 8 + PB_TYPE_FIXEDLEN);
+    sz += put_uint(stats->size()) + stats->size();
+    put_bytes(*stats);
+  }
+
+  // size of the whole row index entry
+  m_buf->data()[lpos] = (uint8_t)(sz + 2);
 }
 
 size_t ProtobufWriter::write(const PostScript& s)

cpp/src/io/orc/orc.h

@@ -470,16 +470,22 @@ class ProtobufWriter {
  public:
   ProtobufWriter() { m_buf = nullptr; }
   ProtobufWriter(std::vector<uint8_t>* output) { m_buf = output; }
-  void putb(uint8_t v) { m_buf->push_back(v); }
+  void put_byte(uint8_t v) { m_buf->push_back(v); }
+  template <typename T>
+  void put_bytes(T const& values)
+  {
+    m_buf->reserve(m_buf->size() + values.size());
+    m_buf->insert(m_buf->end(), values.cbegin(), values.cend());
+  }
   uint32_t put_uint(uint64_t v)
   {
     int l = 1;
     while (v > 0x7f) {
-      putb(static_cast<uint8_t>(v | 0x80));
+      put_byte(static_cast<uint8_t>(v | 0x80));
       v >>= 7;
       l++;
     }
-    putb(static_cast<uint8_t>(v));
+    put_byte(static_cast<uint8_t>(v));
     return l;
   }
   uint32_t put_int(int64_t v)
@@ -493,7 +499,8 @@ class ProtobufWriter {
                            int32_t data_ofs,
                            int32_t data2_blk,
                            int32_t data2_ofs,
-                           TypeKind kind);
+                           TypeKind kind,
+                           ColStatsBlob const* stats);
 
  public:
   size_t write(const PostScript&);

cpp/src/io/orc/orc_field_writer.hpp

@@ -54,7 +54,7 @@ struct ProtobufWriter::ProtobufFieldWriter {
   {
     struct_size += p->put_uint(field * 8 + PB_TYPE_FIXEDLEN);
     auto lpos = p->m_buf->size();
-    p->putb(0);
+    p->put_byte(0);
     auto sz = std::accumulate(value.begin(), value.end(), 0, [p = this->p](size_t sum, auto val) {
       return sum + p->put_uint(val);
     });
@@ -73,8 +73,7 @@ struct ProtobufWriter::ProtobufFieldWriter {
     size_t len = value.length();
     struct_size += p->put_uint(field * 8 + PB_TYPE_FIXEDLEN);
     struct_size += p->put_uint(len) + len;
-    for (size_t i = 0; i < len; i++)
-      p->putb(value[i]);
+    p->put_bytes(value);
   }
 
   /**
@@ -86,8 +85,7 @@ struct ProtobufWriter::ProtobufFieldWriter {
     size_t len = value.size();
     struct_size += p->put_uint(field * 8 + PB_TYPE_FIXEDLEN);
     struct_size += p->put_uint(len) + len;
-    for (size_t i = 0; i < len; i++)
-      p->putb(value[i]);
+    p->put_bytes(value);
   }
 
   /**
@@ -98,7 +96,7 @@ struct ProtobufWriter::ProtobufFieldWriter {
   {
     struct_size += p->put_uint((field)*8 + PB_TYPE_FIXEDLEN);
     auto lpos = p->m_buf->size();
-    p->putb(0);
+    p->put_byte(0);
     auto sz = p->write(value);
     struct_size += sz + 1;
     for (; sz > 0x7f; sz >>= 7, struct_size++)