Add benchmarks for fused DecodeAndCount optimization

Add ReadLevels_RleCountSeparate and ReadLevels_RleCountFused benchmarks
to compare the old approach (Decode + std::count) vs the new fused
DecodeAndCount approach.

Results show ~12% speedup for RLE-heavy data (high repeat counts)
where counting is O(1) for entire runs.

Author: tennisleng

cpp/src/parquet/column_reader_benchmark.cc

@@ -336,6 +336,77 @@ static void DecodeLevels(Encoding::type level_encoding, int16_t max_level, int n
   state.SetItemsProcessed(state.iterations() * num_levels);
 }
 
+// Benchmark that simulates the old approach: Decode + separate count
+static void DecodeLevelsAndCountSeparate(Encoding::type level_encoding, int16_t max_level,
+                                         int num_levels, int batch_size,
+                                         int level_repeat_count,
+                                         ::benchmark::State& state) {
+  std::vector<uint8_t> bytes;
+  {
+    std::vector<int16_t> input_levels;
+    GenerateLevels(/*level_repeats=*/level_repeat_count, /*max_repeat_factor=*/max_level,
+                   num_levels, &input_levels);
+    EncodeLevels(level_encoding, max_level, num_levels, input_levels.data(), &bytes);
+  }
+
+  LevelDecoder decoder;
+  std::vector<int16_t> output_levels(batch_size);
+  for (auto _ : state) {
+    state.PauseTiming();
+    decoder.SetData(level_encoding, max_level, num_levels, bytes.data(),
+                    static_cast<int>(bytes.size()));
+    int64_t total_count = 0;
+    state.ResumeTiming();
+    // Decode + count separately (old approach)
+    while (true) {
+      int levels_decoded = decoder.Decode(batch_size, output_levels.data());
+      if (levels_decoded == 0) {
+        break;
+      }
+      // Separate count pass (simulating the old approach)
+      total_count +=
+          std::count(output_levels.data(), output_levels.data() + levels_decoded, max_level);
+    }
+    DoNotOptimize(total_count);
+  }
+  state.SetBytesProcessed(state.iterations() * num_levels * sizeof(int16_t));
+  state.SetItemsProcessed(state.iterations() * num_levels);
+}
+
+// Benchmark that uses the new fused DecodeAndCount approach
+static void DecodeLevelsAndCountFused(Encoding::type level_encoding, int16_t max_level,
+                                      int num_levels, int batch_size,
+                                      int level_repeat_count, ::benchmark::State& state) {
+  std::vector<uint8_t> bytes;
+  {
+    std::vector<int16_t> input_levels;
+    GenerateLevels(/*level_repeats=*/level_repeat_count, /*max_repeat_factor=*/max_level,
+                   num_levels, &input_levels);
+    EncodeLevels(level_encoding, max_level, num_levels, input_levels.data(), &bytes);
+  }
+
+  LevelDecoder decoder;
+  std::vector<int16_t> output_levels(batch_size);
+  for (auto _ : state) {
+    state.PauseTiming();
+    decoder.SetData(level_encoding, max_level, num_levels, bytes.data(),
+                    static_cast<int>(bytes.size()));
+    int64_t total_count = 0;
+    state.ResumeTiming();
+    // Fused decode + count (new approach)
+    while (true) {
+      int levels_decoded =
+          decoder.DecodeAndCount(batch_size, output_levels.data(), &total_count);
+      if (levels_decoded == 0) {
+        break;
+      }
+    }
+    DoNotOptimize(total_count);
+  }
+  state.SetBytesProcessed(state.iterations() * num_levels * sizeof(int16_t));
+  state.SetItemsProcessed(state.iterations() * num_levels);
+}
+
 static void ReadLevels_Rle(::benchmark::State& state) {
   int16_t max_level = static_cast<int16_t>(state.range(0));
   int num_levels = static_cast<int>(state.range(1));
@@ -354,6 +425,26 @@ static void ReadLevels_BitPack(::benchmark::State& state) {
                level_repeat_count, state);
 }
 
+// Benchmark: Decode + Count separately (old approach)
+static void ReadLevels_RleCountSeparate(::benchmark::State& state) {
+  int16_t max_level = static_cast<int16_t>(state.range(0));
+  int num_levels = static_cast<int>(state.range(1));
+  int batch_size = static_cast<int>(state.range(2));
+  int level_repeat_count = static_cast<int>(state.range(3));
+  DecodeLevelsAndCountSeparate(Encoding::RLE, max_level, num_levels, batch_size,
+                               level_repeat_count, state);
+}
+
+// Benchmark: Fused DecodeAndCount (new approach)
+static void ReadLevels_RleCountFused(::benchmark::State& state) {
+  int16_t max_level = static_cast<int16_t>(state.range(0));
+  int num_levels = static_cast<int>(state.range(1));
+  int batch_size = static_cast<int>(state.range(2));
+  int level_repeat_count = static_cast<int>(state.range(3));
+  DecodeLevelsAndCountFused(Encoding::RLE, max_level, num_levels, batch_size,
+                            level_repeat_count, state);
+}
+
 static void ReadLevelsArguments(::benchmark::internal::Benchmark* b) {
   b->ArgNames({"MaxLevel", "NumLevels", "BatchSize", "LevelRepeatCount"})
       ->Args({1, 8096, 1024, 1})
@@ -367,6 +458,8 @@ static void ReadLevelsArguments(::benchmark::internal::Benchmark* b) {
 
 BENCHMARK(ReadLevels_Rle)->Apply(ReadLevelsArguments);
 BENCHMARK(ReadLevels_BitPack)->Apply(ReadLevelsArguments);
+BENCHMARK(ReadLevels_RleCountSeparate)->Apply(ReadLevelsArguments);
+BENCHMARK(ReadLevels_RleCountFused)->Apply(ReadLevelsArguments);
 
 }  // namespace benchmarks
 }  // namespace parquet