DO NOT MERGE: EXP: [C++] VisitTwoBitRuns changes

cpp/src/arrow/compute/kernels/codegen_internal.h

@@ -38,6 +38,7 @@
 #include "arrow/type.h"
 #include "arrow/type_traits.h"
 #include "arrow/util/bit_block_counter.h"
+#include "arrow/util/bit_run_reader.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/bitmap_generate.h"
 #include "arrow/util/bitmap_reader.h"
@@ -509,18 +510,24 @@ static void VisitTwoArrayValuesInline(const ArraySpan& arr0, const ArraySpan& ar
   ArrayIterator<Arg0Type> arr0_it(arr0);
   ArrayIterator<Arg1Type> arr1_it(arr1);
 
-  auto visit_valid = [&](int64_t i) {
-    valid_func(GetViewType<Arg0Type>::LogicalValue(arr0_it()),
-               GetViewType<Arg1Type>::LogicalValue(arr1_it()));
-  };
-  auto visit_null = [&]() {
-    arr0_it();
-    arr1_it();
-    null_func();
+  auto visit_run = [&](int64_t position, int64_t run_length, bool is_set) {
+    if (is_set) {
+      for (int64_t i = 0; i < run_length; ++i) {
+        valid_func(GetViewType<Arg0Type>::LogicalValue(arr0_it()),
+                   GetViewType<Arg1Type>::LogicalValue(arr1_it()));
+      }
+    } else {
+      for (int64_t i = 0; i < run_length; ++i) {
+        arr0_it();
+        arr1_it();
+        null_func();
+      }
+    }
   };
-  VisitTwoBitBlocksVoid(arr0.buffers[0].data, arr0.offset, arr1.buffers[0].data,
-                        arr1.offset, arr0.length, std::move(visit_valid),
-                        std::move(visit_null));
+
+  ::arrow::internal::VisitTwoBitRunsVoid(arr0.buffers[0].data, arr0.offset,
+                                         arr1.buffers[0].data, arr1.offset, arr0.length,
+                                         std::move(visit_run));
 }
 
 // ----------------------------------------------------------------------

cpp/src/arrow/compute/kernels/scalar_if_else.cc

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <cstdint>
 #include <cstring>
 #include <limits>
 #include "arrow/array/builder_nested.h"
@@ -1607,26 +1608,14 @@ Status ExecArrayCaseWhen(KernelContext* ctx, const ExecSpan& batch, ExecResult*
 
     if (cond_array.GetNullCount() == 0) {
       // If no valid buffer, visit mask & cond bitmap simultaneously
-      BinaryBitBlockCounter counter(mask, /*start_offset=*/0, cond_values, cond_offset,
-                                    batch.length);
-      while (offset < batch.length) {
-        const auto block = counter.NextAndWord();
-        if (block.AllSet()) {
-          CopyValues<Type>(value, offset, block.length, out_valid, out_values,
-                           out_offset + offset);
-          bit_util::SetBitsTo(mask, offset, block.length, false);
-        } else if (block.popcount) {
-          for (int64_t j = 0; j < block.length; ++j) {
-            if (bit_util::GetBit(mask, offset + j) &&
-                bit_util::GetBit(cond_values, cond_offset + offset + j)) {
-              CopyValues<Type>(value, offset + j, /*length=*/1, out_valid, out_values,
-                               out_offset + offset + j);
-              bit_util::SetBitTo(mask, offset + j, false);
-            }
-          }
-        }
-        offset += block.length;
-      }
+      auto visit_run = [&](int64_t position, int64_t run_length) {
+        CopyValues<Type>(value, position, run_length, out_valid, out_values,
+                         out_offset + position);
+        bit_util::SetBitsTo(mask, position, run_length, false);
+        return Status::OK();
+      };
+      RETURN_NOT_OK(::arrow::internal::VisitTwoSetBitRuns(
+          mask, /*start_offset=*/0, cond_values, cond_offset, batch.length, visit_run));
     } else {
       // Visit mask & cond bitmap & cond validity
       const uint8_t* cond_valid = cond_array.buffers[0].data;
@@ -1657,32 +1646,20 @@ Status ExecArrayCaseWhen(KernelContext* ctx, const ExecSpan& batch, ExecResult*
   }
   if (!have_else_arg) {
     // Need to initialize any remaining null slots (uninitialized memory)
-    BitBlockCounter counter(mask, /*offset=*/0, batch.length);
-    int64_t offset = 0;
     auto bit_width = checked_cast<const FixedWidthType&>(*out->type()).bit_width();
     auto byte_width = bit_util::BytesForBits(bit_width);
-    while (offset < batch.length) {
-      const auto block = counter.NextWord();
-      if (block.AllSet()) {
-        if (bit_width == 1) {
-          bit_util::SetBitsTo(out_values, out_offset + offset, block.length, false);
-        } else {
-          std::memset(out_values + (out_offset + offset) * byte_width, 0x00,
-                      byte_width * block.length);
-        }
-      } else if (!block.NoneSet()) {
-        for (int64_t j = 0; j < block.length; ++j) {
-          if (bit_util::GetBit(out_valid, out_offset + offset + j)) continue;
-          if (bit_width == 1) {
-            bit_util::ClearBit(out_values, out_offset + offset + j);
-          } else {
-            std::memset(out_values + (out_offset + offset + j) * byte_width, 0x00,
-                        byte_width);
-          }
-        }
+
+    auto visit_run = [&](int64_t position, int64_t run_length) {
+      if (bit_width == 1) {
+        bit_util::SetBitsTo(out_values, out_offset + position, run_length, false);
+      } else {
+        std::memset(out_values + (out_offset + position) * byte_width, 0x00,
+                    byte_width * run_length);
       }
-      offset += block.length;
-    }
+      bit_util::SetBitsTo(mask, position, run_length, false);
+    };
+
+    ::arrow::internal::VisitSetBitRunsVoid(mask, /*offset=*/0, batch.length, visit_run);
   }
   return Status::OK();
 }

cpp/src/arrow/compute/kernels/vector_selection_take_internal.cc

@@ -34,6 +34,7 @@
 #include "arrow/compute/registry.h"
 #include "arrow/memory_pool.h"
 #include "arrow/record_batch.h"
+#include "arrow/status.h"
 #include "arrow/table.h"
 #include "arrow/type.h"
 #include "arrow/type_traits.h"
@@ -135,37 +136,17 @@ Result<std::shared_ptr<ArrayData>> GetTakeIndicesFromBitmapImpl(
     // data bitmap together.
     DCHECK_EQ(null_selection, FilterOptions::DROP);
 
-    // The position relative to the start of the filter
-    T position = 0;
-    // The current position taking the filter offset into account
-    int64_t position_with_offset = filter.offset;
-
-    BinaryBitBlockCounter filter_counter(filter_data, filter.offset, filter_is_valid,
-                                         filter.offset, filter.length);
-    while (position < filter.length) {
-      BitBlockCount and_block = filter_counter.NextAndWord();
-      RETURN_NOT_OK(builder.Reserve(and_block.popcount));
-      if (and_block.AllSet()) {
-        // All the values are selected and non-null
-        for (int64_t i = 0; i < and_block.length; ++i) {
-          builder.UnsafeAppend(position++);
-        }
-        position_with_offset += and_block.length;
-      } else if (!and_block.NoneSet()) {
-        // Some of the values are false or null
-        for (int64_t i = 0; i < and_block.length; ++i) {
-          if (bit_util::GetBit(filter_is_valid, position_with_offset) &&
-              bit_util::GetBit(filter_data, position_with_offset)) {
-            builder.UnsafeAppend(position);
-          }
-          ++position;
-          ++position_with_offset;
-        }
-      } else {
-        position += and_block.length;
-        position_with_offset += and_block.length;
+    auto visit_run = [&](int64_t position, int64_t run_length) {
+      RETURN_NOT_OK(builder.Reserve(run_length));
+      for (int64_t i = 0; i < run_length; ++i) {
+        builder.UnsafeAppend(static_cast<T>(position + i));
       }
-    }
+      return Status::OK();
+    };
+
+    RETURN_NOT_OK(::arrow::internal::VisitTwoSetBitRuns(filter_data, filter.offset,
+                                                        filter_is_valid, filter.offset,
+                                                        filter.length, visit_run));
   } else {
     // The filter has no nulls, so we need only look for true values
     RETURN_NOT_OK(::arrow::internal::VisitSetBitRuns(

cpp/src/arrow/util/CMakeLists.txt

@@ -128,6 +128,7 @@ add_arrow_benchmark(decimal_benchmark)
 add_arrow_benchmark(hashing_benchmark)
 add_arrow_benchmark(int_util_benchmark)
 add_arrow_benchmark(machine_benchmark)
+add_arrow_benchmark(nested_bitmap_traversal_benchmark)
 add_arrow_benchmark(queue_benchmark)
 add_arrow_benchmark(range_benchmark)
 add_arrow_benchmark(small_vector_benchmark)