GH-55399: [C++][Parquet] Use CountIf16 for level counting in reader and writer

### Rationale for this change

Follow-up to the `CountIf16` helper: use it for the level counting loops
in the Parquet column reader/writer and in the file scanner, replacing
open-coded scalar loops and `std::count` calls.

Converted sites:

- `TypedColumnReaderImpl::ReadLevelsInCurrentPage`: count non-null values.
- `TypedRecordReader::SkipRecordsInBufferNonRepeated`: count values to skip.
- `TypedRecordReader::DelimitRecords`: count records (rep level == 0) and
  values seen (def level == max).
- `TypedRecordReader::ReadDenseForOptionalInBuffer`: count values to read.
- `TypedColumnWriterImpl::WriteLevels`: count values to write (both
  overloads) and records written (rep level == 0).
- `TypedColumnWriterImpl::MaybeCalculateValidityBits`: count values to
  write and spaced values to write.
- `ScanFileContents`: count rows (rep level == 0).

Sites are left as is where counting is not the bottleneck: the
per-element bitmap loops in `DefRepLevelsToListInfo` also build validity
bitmaps and write list offsets for every level, `page_index.cc` counts
pages rather than levels, and the non-nullable validation loop in
`WriteArrow` needs to `break` early on the first mismatch.

`UpdateLevelHistogram`'s `max_level == 1` fast path is also left as is: its
levels are known to be 0 or 1, so it sums them with `std::accumulate` in the
`int16_t` space, which is a single 16-bit add per vector, while counting the
1s costs a compare and a subtract.

### What changes are included in this PR?

Use `CountIf16` for the level counting call sites above.

### Are these changes tested?

Covered by the existing Parquet reader, writer and Arrow reader/writer
tests.

### Are there any user-facing changes?

No.

Author: mapleFU

cpp/src/parquet/column_reader.cc

@@ -47,6 +47,7 @@
 #include "parquet/encryption/encryption_internal.h"
 #include "parquet/encryption/internal_file_decryptor.h"
 #include "parquet/exception.h"
+#include "parquet/level_comparison.h"
 #include "parquet/level_conversion.h"
 #include "parquet/level_decoder_internal.h"
 #include "parquet/properties.h"
@@ -1546,8 +1547,10 @@ class TypedColumnReaderImpl
       // TODO(wesm): this tallying of values-to-decode can be performed with better
       // cache-efficiency if fused with the level decoding.
       ARROW_DCHECK_LE(*num_def_levels, std::numeric_limits<int32_t>::max());
-      *non_null_values_to_read += static_cast<int32_t>(
-          std::count(def_levels, def_levels + *num_def_levels, this->max_def_level()));
+      const int16_t max_def_level = this->max_def_level();
+      *non_null_values_to_read += static_cast<int32_t>(internal::CountIf16(
+          def_levels, *num_def_levels,
+          [max_def_level](int16_t level) { return level == max_def_level; }));
     } else {
       // Required field, read all values
       if (num_def_levels != nullptr) {
@@ -1961,9 +1964,10 @@ int64_t TypedRecordReader<DT, VS>::SkipRecordsInBufferNonRepeated(int64_t num_re
   // We skipped the levels by incrementing 'levels_position_'. For values
   // we do not have a buffer, so we need to read them and throw them away.
   // First we need to figure out how many present/not-null values there are.
-  const auto values_to_read = static_cast<int32_t>(
-      std::count(def_levels() + levels_position_, def_levels() + remaining_levels_pos,
-                 this->max_def_level()));
+  const int16_t max_def_level = this->max_def_level();
+  const auto values_to_read = static_cast<int32_t>(internal::CountIf16(
+      def_levels() + levels_position_, remaining_levels_pos - levels_position_,
+      [max_def_level](int16_t level) { return level == max_def_level; }));
 
   // Now that we have figured out number of values to read, we do not need
   // these levels anymore. We will remove these values from the buffer.
@@ -2128,9 +2132,8 @@ int64_t TypedRecordReader<DT, VS>::DelimitRecords(int64_t num_records,
     const int64_t stride =
         std::min(levels_written() - levels_position_, num_records - records_read);
     const int64_t position_end = levels_position_ + stride;
-    for (int64_t i = levels_position_; i < position_end; ++i) {
-      records_read += rep_levels[i] == 0;
-    }
+    records_read += CountIf16(rep_levels + levels_position_, stride,
+                              [](int16_t rep_level) { return rep_level == 0; });
     levels_position_ = position_end;
     if (records_read == num_records) {
       // Check last rep_level reaches the boundary and
@@ -2144,8 +2147,10 @@ int64_t TypedRecordReader<DT, VS>::DelimitRecords(int64_t num_records,
     }
   }
   // Scan definition levels to find number of physical values
-  *values_seen = std::count(def_levels + level, def_levels + levels_position_,
-                            this->max_def_level());
+  const int16_t max_def_level = this->max_def_level();
+  *values_seen = CountIf16(
+      def_levels + level, levels_position_ - level,
+      [max_def_level](int16_t def_level) { return def_level == max_def_level; });
   return records_read;
 }
 
@@ -2234,8 +2239,10 @@ void TypedRecordReader<DT, VS>::ReadDenseForOptionalInBuffer(
 
   // When reading dense we need to figure out number of values to read.
   const int16_t* def_levels = this->def_levels();
-  *values_to_read += std::count(def_levels + start_levels_position,
-                                def_levels + levels_position_, this->max_def_level());
+  const int16_t max_def_level = this->max_def_level();
+  *values_to_read += CountIf16(
+      def_levels + start_levels_position, levels_position_ - start_levels_position,
+      [max_def_level](int16_t def_level) { return def_level == max_def_level; });
   // This is only reading in the current page so this fits in an int32.
   value_sink_.ReadValuesDense(*this->current_decoder_.get(),
                               clamp_to<int32_t>(*values_to_read));

cpp/src/parquet/column_writer.cc

@@ -55,6 +55,7 @@
 #include "parquet/encoding.h"
 #include "parquet/encryption/encryption_internal.h"
 #include "parquet/encryption/internal_file_encryptor.h"
+#include "parquet/level_comparison.h"
 #include "parquet/level_conversion.h"
 #include "parquet/metadata.h"
 #include "parquet/page_index.h"
@@ -1698,11 +1699,10 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
     int64_t values_to_write = 0;
     // If the field is required and non-repeated, there are no definition levels
     if (descr_->max_definition_level() > 0) {
-      for (int64_t i = 0; i < num_levels; ++i) {
-        if (def_levels[i] == descr_->max_definition_level()) {
-          ++values_to_write;
-        }
-      }
+      const int16_t max_def_level = descr_->max_definition_level();
+      values_to_write = internal::CountIf16(
+          def_levels, num_levels,
+          [max_def_level](int16_t def_level) { return def_level == max_def_level; });
 
       WriteDefinitionLevels(num_levels, def_levels);
     } else {
@@ -1714,12 +1714,10 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
     if (descr_->max_repetition_level() > 0) {
       // A row could include more than one value
       // Count the occasions where we start a new row
-      for (int64_t i = 0; i < num_levels; ++i) {
-        if (rep_levels[i] == 0) {
-          rows_written_++;
-          num_buffered_rows_++;
-        }
-      }
+      const int64_t records_written = internal::CountIf16(
+          rep_levels, num_levels, [](int16_t rep_level) { return rep_level == 0; });
+      rows_written_ += records_written;
+      num_buffered_rows_ += records_written;
 
       WriteRepetitionLevels(num_levels, rep_levels);
     } else {
@@ -1749,11 +1747,16 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
         *out_spaced_values_to_write = batch_size;
         *null_count = 0;
       } else {
-        for (int x = 0; x < batch_size; x++) {
-          *out_values_to_write += def_levels[x] == level_info_.def_level ? 1 : 0;
-          *out_spaced_values_to_write +=
-              def_levels[x] >= level_info_.repeated_ancestor_def_level ? 1 : 0;
-        }
+        const int16_t def_level = level_info_.def_level;
+        const int16_t repeated_ancestor_def_level =
+            level_info_.repeated_ancestor_def_level;
+        *out_values_to_write += internal::CountIf16(
+            def_levels, batch_size,
+            [def_level](int16_t level) { return level == def_level; });
+        *out_spaced_values_to_write += internal::CountIf16(
+            def_levels, batch_size, [repeated_ancestor_def_level](int16_t level) {
+              return level >= repeated_ancestor_def_level;
+            });
         *null_count = batch_size - *out_values_to_write;
       }
       return;
@@ -1809,12 +1812,10 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
     if (descr_->max_repetition_level() > 0) {
       // A row could include more than one value
       // Count the occasions where we start a new row
-      for (int64_t i = 0; i < num_levels; ++i) {
-        if (rep_levels[i] == 0) {
-          rows_written_++;
-          num_buffered_rows_++;
-        }
-      }
+      const int64_t records_written = internal::CountIf16(
+          rep_levels, num_levels, [](int16_t rep_level) { return rep_level == 0; });
+      rows_written_ += records_written;
+      num_buffered_rows_ += records_written;
       WriteRepetitionLevels(num_levels, rep_levels);
     } else {
       // Each value is exactly one row

cpp/src/parquet/file_reader.cc

@@ -44,6 +44,7 @@
 #include "parquet/encryption/internal_file_decryptor.h"
 #include "parquet/exception.h"
 #include "parquet/file_writer.h"
+#include "parquet/level_comparison.h"
 #include "parquet/metadata.h"
 #include "parquet/page_index.h"
 #include "parquet/platform.h"
@@ -977,11 +978,9 @@ int64_t ScanFileContents(std::vector<int> columns, const int32_t column_batch_si
             ScanAllValues(column_batch_size, def_levels.data(), rep_levels.data(),
                           values.data(), &values_read, col_reader.get());
         if (col_reader->descr()->max_repetition_level() > 0) {
-          for (size_t i = 0; i < static_cast<size_t>(levels_read); i++) {
-            if (rep_levels[i] == 0) {
-              total_rows[col]++;
-            }
-          }
+          total_rows[col] +=
+              internal::CountIf16(rep_levels.data(), levels_read,
+                                  [](int16_t rep_level) { return rep_level == 0; });
         } else {
           total_rows[col] += levels_read;
         }

cpp/src/parquet/size_statistics.cc

@@ -19,14 +19,14 @@
 
 #include <algorithm>
 #include <array>
-#include <numeric>
 #include <ostream>
 #include <span>
 #include <string_view>
 #include <vector>
 
 #include "arrow/util/logging_internal.h"
 #include "parquet/exception.h"
+#include "parquet/level_comparison.h"
 #include "parquet/schema.h"
 
 namespace parquet {
@@ -160,17 +160,11 @@ void UpdateLevelHistogram(std::span<const int16_t> levels, std::span<int64_t> hi
 
   if (max_level == 1) {
     // Specialize the common case for non-repeated non-nested columns.
-    // Summing the levels gives us the number of 1s, and the number of 0s follows.
-    // We do repeated sums in the int16_t space, which the compiler is likely
-    // to vectorize efficiently.
-    constexpr int64_t kChunkSize = 1 << 14;  // to avoid int16_t overflows
-    int64_t hist1 = 0;
-    auto it = levels.begin();
-    while (it != levels.end()) {
-      const auto chunk_size = std::min<int64_t>(levels.end() - it, kChunkSize);
-      hist1 += std::accumulate(it, it + chunk_size, int16_t{0});
-      it += chunk_size;
-    }
+    // Counting the 1s directly gives us the number of 1s, and the number of 0s
+    // follows. CountIf16 accumulates in the int16_t space in chunks, which the
+    // compiler vectorizes efficiently and which cannot overflow.
+    const int64_t hist1 = internal::CountIf16(levels.data(), num_levels,
+                                              [](int16_t level) { return level == 1; });
     histogram[0] += num_levels - hist1;
     histogram[1] += hist1;
     return;