Reuse Buffer for each recordbatch

cpp/CMakeLists.txt

@@ -48,6 +48,7 @@ if(POLICY CMP0074)
 endif()
 
 set(ARROW_VERSION "4.0.0")
+#add_compile_options(-g -O0)
 
 string(REGEX MATCH "^[0-9]+\\.[0-9]+\\.[0-9]+" ARROW_BASE_VERSION "${ARROW_VERSION}")
 
@@ -937,3 +938,7 @@ config_summary_message()
 if(${ARROW_BUILD_CONFIG_SUMMARY_JSON})
   config_summary_json()
 endif()
+
+
+
+

cpp/README.md

@@ -32,3 +32,17 @@ to install pre-compiled binary versions of the library.
 Please refer to our latest [C++ Development Documentation][1].
 
 [1]: https://github.com/apache/arrow/blob/master/docs/source/developers/cpp
+
+## Run parquet string scan benchmark
+#### Minimal benchmark build
+cd arrow
+mkdir -p cpp/debug
+cd cpp/debug
+cmake -DCMAKE_BUILD_TYPE=Release -DARROW_BUILD_BENCHMARKS=ON -DARROW_WITH_ZLIB=ON -DARROW_JEMALLOC=OFF -DARROW_PARQUET=ON -DARROW_COMPUTE=ON -DARROW_DATASET=ON -DARROW_WITH_SNAPPY=ON -DARROW_FILESYSTEM=ON ..
+
+#### Run benchmark and collect perf data
+cpp/debug
+./release/parquet-arrow-parquet-scan-string-benchmark --iterations 10  --threads 1  --file {parquet_path} --cpu 0 &
+perf record -e cycles:ppp -C 0 sleep 10
+perf report
+

cpp/src/parquet/CMakeLists.txt

@@ -399,6 +399,8 @@ add_parquet_benchmark(column_io_benchmark)
 add_parquet_benchmark(encoding_benchmark)
 add_parquet_benchmark(level_conversion_benchmark)
 add_parquet_benchmark(arrow/reader_writer_benchmark PREFIX "parquet-arrow")
+add_parquet_benchmark(arrow/parquet_scan_benchmark PREFIX "parquet-arrow")
+add_parquet_benchmark(arrow/parquet_scan_string_benchmark PREFIX "parquet-arrow")
 
 if(ARROW_WITH_BROTLI)
   add_definitions(-DARROW_WITH_BROTLI)

cpp/src/parquet/arrow/parquet_scan_benchmark.cc

@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <arrow/filesystem/filesystem.h>
+#include <arrow/io/interfaces.h>
+#include <arrow/memory_pool.h>
+#include <arrow/record_batch.h>
+#include <arrow/testing/gtest_util.h>
+#include <arrow/type.h>
+#include <arrow/util/io_util.h>
+#include <benchmark/benchmark.h>
+#include <gtest/gtest.h>
+#include <parquet/arrow/reader.h>
+#include <parquet/file_reader.h>
+
+#include <chrono>
+
+#include "arrow/record_batch.h"
+#include "parquet/arrow/utils/macros.h"
+#include "parquet/arrow/test_utils.h"
+
+
+// namespace parquet {
+// namespace benchmark {
+
+const int batch_buffer_size = 32768;
+
+class GoogleBenchmarkColumnarToRow {
+ public:
+  GoogleBenchmarkColumnarToRow(std::string file_name) { GetRecordBatchReader(file_name); }
+
+  void GetRecordBatchReader(const std::string& input_file) {
+    std::unique_ptr<::parquet::arrow::FileReader> parquet_reader;
+    std::shared_ptr<arrow::RecordBatchReader> record_batch_reader;
+
+    std::shared_ptr<arrow::fs::FileSystem> fs;
+    std::string file_name;
+    ARROW_ASSIGN_OR_THROW(fs, arrow::fs::FileSystemFromUriOrPath(input_file, &file_name))
+
+    ARROW_ASSIGN_OR_THROW(file, fs->OpenInputFile(file_name));
+
+    properties.set_batch_size(batch_buffer_size);
+    properties.set_pre_buffer(false);
+    properties.set_use_threads(false);
+
+    ASSERT_NOT_OK(::parquet::arrow::FileReader::Make(
+        arrow::default_memory_pool(), ::parquet::ParquetFileReader::Open(file),
+        properties, &parquet_reader));
+
+    ASSERT_NOT_OK(parquet_reader->GetSchema(&schema));
+
+    auto num_rowgroups = parquet_reader->num_row_groups();
+
+    for (int i = 0; i < num_rowgroups; ++i) {
+      row_group_indices.push_back(i);
+    }
+
+    auto num_columns = schema->num_fields();
+    for (int i = 0; i < num_columns; ++i) {
+      column_indices.push_back(i);
+    }
+  }
+
+  virtual void operator()(benchmark::State& state) {}
+
+ protected:
+  long SetCPU(uint32_t cpuindex) {
+    cpu_set_t cs;
+    CPU_ZERO(&cs);
+    CPU_SET(cpuindex, &cs);
+    return sched_setaffinity(0, sizeof(cs), &cs);
+  }
+
+ protected:
+  std::string file_name;
+  std::shared_ptr<arrow::io::RandomAccessFile> file;
+  std::vector<int> row_group_indices;
+  std::vector<int> column_indices;
+  std::shared_ptr<arrow::Schema> schema;
+  parquet::ArrowReaderProperties properties;
+};
+class GoogleBenchmarkColumnarToRow_CacheScan_Benchmark
+    : public GoogleBenchmarkColumnarToRow {
+ public:
+  GoogleBenchmarkColumnarToRow_CacheScan_Benchmark(std::string filename)
+      : GoogleBenchmarkColumnarToRow(filename) {}
+  void operator()(benchmark::State& state) {
+    if (state.range(0) == 0xffffffff) {
+      SetCPU(state.thread_index());
+    } else {
+      SetCPU(state.range(0));
+    }
+
+    arrow::Compression::type compression_type = (arrow::Compression::type)1;
+
+    std::shared_ptr<arrow::RecordBatch> record_batch;
+    int64_t elapse_read = 0;
+    int64_t num_batches = 0;
+    int64_t num_rows = 0;
+    int64_t init_time = 0;
+    int64_t write_time = 0;
+
+
+    std::vector<int> local_column_indices = column_indices;
+
+    std::shared_ptr<arrow::Schema> local_schema;
+    local_schema = std::make_shared<arrow::Schema>(*schema.get());
+
+    if (state.thread_index() == 0) std::cout << local_schema->ToString() << std::endl;
+
+    for (auto _ : state) {
+      std::unique_ptr<::parquet::arrow::FileReader> parquet_reader;
+      std::shared_ptr<arrow::RecordBatchReader> record_batch_reader;
+      ASSERT_NOT_OK(::parquet::arrow::FileReader::Make(
+          ::arrow::default_memory_pool(), ::parquet::ParquetFileReader::Open(file),
+          properties, &parquet_reader));
+
+      std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+      ASSERT_NOT_OK(parquet_reader->GetRecordBatchReader(
+          row_group_indices, local_column_indices, &record_batch_reader));
+      do {
+        TIME_NANO_OR_THROW(elapse_read, record_batch_reader->ReadNext(&record_batch));
+
+        if (record_batch) {
+          // batches.push_back(record_batch);
+          num_batches += 1;
+          num_rows += record_batch->num_rows();
+        }
+      } while (record_batch);
+
+      std::cout << " parquet parse done elapsed time = " << elapse_read / 1000000
+              << " rows = " << num_rows << std::endl;
+    }
+
+    state.counters["rowgroups"] =
+        benchmark::Counter(row_group_indices.size(), benchmark::Counter::kAvgThreads,
+                           benchmark::Counter::OneK::kIs1000);
+    state.counters["columns"] =
+        benchmark::Counter(column_indices.size(), benchmark::Counter::kAvgThreads,
+                           benchmark::Counter::OneK::kIs1000);
+    state.counters["batches"] = benchmark::Counter(
+        num_batches, benchmark::Counter::kAvgThreads, benchmark::Counter::OneK::kIs1000);
+    state.counters["num_rows"] = benchmark::Counter(
+        num_rows, benchmark::Counter::kAvgThreads, benchmark::Counter::OneK::kIs1000);
+    state.counters["batch_buffer_size"] =
+        benchmark::Counter(batch_buffer_size, benchmark::Counter::kAvgThreads,
+                           benchmark::Counter::OneK::kIs1024);
+
+    state.counters["parquet_parse"] = benchmark::Counter(
+        elapse_read, benchmark::Counter::kAvgThreads, benchmark::Counter::OneK::kIs1000);
+    state.counters["init_time"] = benchmark::Counter(
+        init_time, benchmark::Counter::kAvgThreads, benchmark::Counter::OneK::kIs1000);
+    state.counters["write_time"] = benchmark::Counter(
+        write_time, benchmark::Counter::kAvgThreads, benchmark::Counter::OneK::kIs1000);
+  }
+};
+
+// }  // namespace columnartorow
+// }  // namespace sparkcolumnarplugin
+
+int main(int argc, char** argv) {
+  uint32_t iterations = 1;
+  uint32_t threads = 1;
+  std::string datafile;
+  uint32_t cpu = 0xffffffff;
+
+  for (int i = 0; i < argc; i++) {
+    if (strcmp(argv[i], "--iterations") == 0) {
+      iterations = atol(argv[i + 1]);
+    } else if (strcmp(argv[i], "--threads") == 0) {
+      threads = atol(argv[i + 1]);
+    } else if (strcmp(argv[i], "--file") == 0) {
+      datafile = argv[i + 1];
+    } else if (strcmp(argv[i], "--cpu") == 0) {
+      cpu = atol(argv[i + 1]);
+    }
+  }
+  std::cout << "iterations = " << iterations << std::endl;
+  std::cout << "threads = " << threads << std::endl;
+  std::cout << "datafile = " << datafile << std::endl;
+  std::cout << "cpu = " << cpu << std::endl;
+
+  GoogleBenchmarkColumnarToRow_CacheScan_Benchmark
+      bck(datafile);
+
+  benchmark::RegisterBenchmark("GoogleBenchmarkColumnarToRow::CacheScan", bck)
+      ->Args({
+          cpu,
+      })
+      ->Iterations(iterations)
+      ->Threads(threads)
+      ->ReportAggregatesOnly(false)
+      ->MeasureProcessCPUTime()
+      ->Unit(benchmark::kSecond);
+
+  benchmark::Initialize(&argc, argv);
+  benchmark::RunSpecifiedBenchmarks();
+  benchmark::Shutdown();
+}