   return count;
 }
 
+constexpr int BITS_PER_MASK32 = GDF_VALID_BITSIZE * RATIO;
 
 __global__ 
-void count_valid_bits(valid32_t const * const __restrict__ masks32, const int num_masks32, int * const __restrict__ global_count)
+void count_valid_bits(valid32_t const * const __restrict__ masks32, 
+                      const int num_masks32, 
+                      const int num_rows, 
+                      int * const __restrict__ global_count)
 {
   typedef cub::BlockReduce<int, block_size> BlockReduce;
   __shared__ typename BlockReduce::TempStorage temp_storage;

  const size_t num_masks = gdf_get_num_chars_bitmask(num_rows);

  // Number of 4 byte types in the validity bit mask 
  size_t num_masks32 = num_masks / RATIO;

  // If the total number of masks is not a multiple of the RATIO 
  // between the original mask type and the 4 byte masks type, then 
  // these "remainder" masks cannot be proccessed in the transform_reduce
  // and must be handled separately
  cudaStream_t copy_stream;
  size_t num_remainder_masks = num_masks % RATIO;

  // If there are no remainder masks, the last mask still
  // needs to be handled separately
  if(0 == num_remainder_masks){
    num_remainder_masks = RATIO;
    num_masks32--;
  }

  std::vector<gdf_valid_type> remainder_masks(num_remainder_masks); 
  if(remainder_masks.size() > 0)
  {
    CUDA_TRY(cudaStreamCreate(&copy_stream));

    // Copy the remainder masks to the host
    // FIXME: Is this endian safe?
    const gdf_valid_type * first_remainder_mask = &(masks[num_masks - num_remainder_masks]);

    CUDA_TRY( cudaMemcpyAsync(remainder_masks.data(), 
                              first_remainder_mask, 
                              num_remainder_masks * sizeof(gdf_valid_type), 
                              cudaMemcpyDeviceToHost,
                              copy_stream) );
  }

  int * d_count32;
  cudaStream_t count_stream;
  if(num_masks32 > 0)
  {
    CUDA_TRY(cudaStreamCreate(&count_stream));
    // Cast validity buffer to 4 byte type
    valid32_t const * masks32 = reinterpret_cast<valid32_t const *>(masks);

    CUDA_TRY(cudaMalloc(&d_count32, sizeof(int)));
    CUDA_TRY(cudaMemsetAsync(d_count32, 0, sizeof(int),count_stream));

    const int grid_size = (num_masks32 + block_size - 1)/block_size;
    count_valid_bits<<<grid_size, block_size,0,count_stream>>>(masks32, num_masks32, d_count32);

    CUDA_TRY( cudaGetLastError() );
  }

  // Count the number of valid bits in the remainder masks
  size_t remainder_count = 0;
  if(remainder_masks.size() > 0)
  {
    CUDA_TRY(cudaStreamSynchronize(copy_stream));
    CUDA_TRY(cudaStreamDestroy(copy_stream));
    remainder_count = count_valid_bits_host(remainder_masks, num_rows);
  }

  int count32{0};
  if(num_masks32 > 0)
  {
    CUDA_TRY(cudaMemcpyAsync(&count32, d_count32, sizeof(int), cudaMemcpyDeviceToHost,count_stream));
    CUDA_TRY(cudaStreamSynchronize(count_stream));
    CUDA_TRY(cudaStreamDestroy(count_stream));
  }

  // The final count of valid bits is the sum of the result from the
  // transform_reduce and the remainder masks
  *count = (count32 + remainder_count);

  return GDF_SUCCESS;
}

  int cur_mask = threadIdx.x + blockIdx.x * gridDim.x;

  int my_count = 0;
  while(cur_mask < (num_masks32 - 1))
  {
    my_count += __popc(masks32[cur_mask]);

    cur_mask += blockDim.x * gridDim.x;
  }

  if(cur_mask == (num_masks32 - 1))
  {
    valid32_t last_mask = masks32[num_masks32 - 1];
    int num_rows_last_mask = num_rows % BITS_PER_MASK32;
    if(num_rows_last_mask == 0)
      num_rows_last_mask = BITS_PER_MASK32;

    for(int i = 0; i < num_rows_last_mask; ++i)
    {
      my_count += last_mask & gdf_valid_type(1);
      last_mask >>= 1;
    }
  }


  int block_count = BlockReduce(temp_storage).Sum(my_count);

