initialize arrow compute

Author: aalkin

Framework/Core/include/Framework/IndexBuilderHelpers.h

@@ -29,7 +29,7 @@ enum struct IndexKind : int {
 
 namespace o2::framework
 {
-void cannotBuildAnArray();
+void cannotBuildAnArray(const char* reason);
 void cannotCreateIndexBuilder();
 
 struct ChunkedArrayIterator {

Framework/Core/src/IndexBuilderHelpers.cxx

@@ -14,16 +14,17 @@
 #include "Framework/CompilerBuiltins.h"
 #include "Framework/VariantHelpers.h"
 #include <arrow/compute/api_aggregate.h>
+#include <arrow/compute/initialize.h>
 #include <arrow/compute/kernel.h>
 #include <arrow/status.h>
 #include <arrow/table.h>
 #include <arrow/util/key_value_metadata.h>
 
 namespace o2::framework
 {
-void cannotBuildAnArray()
+void cannotBuildAnArray(const char* reason)
 {
-  throw framework::runtime_error("Cannot finish an array");
+  throw framework::runtime_error_f("Cannot finish an array: %s", reason);
 }
 
 void cannotCreateIndexBuilder()
@@ -62,10 +63,10 @@ SelfBuilder::SelfBuilder(arrow::MemoryPool* pool)
 {
   auto status = arrow::MakeBuilder(pool, arrow::int32(), &mBuilder);
   if (!status.ok()) {
-    throw framework::runtime_error("Cannot create array builder for the self-index!");
+    throw framework::runtime_error_f("Cannot create array builder for the self-index: %s", status.ToString().c_str());
   }
 }
-// static_cast<ChunkedArrayIterator*>(this)->reset(pool);
+
 void SelfBuilder::reset(std::shared_ptr<arrow::ChunkedArray>)
 {
   mBuilder->Reset();
@@ -74,15 +75,18 @@ void SelfBuilder::reset(std::shared_ptr<arrow::ChunkedArray>)
 
 void SelfBuilder::fill(int idx)
 {
-  (void)static_cast<arrow::Int32Builder*>(mBuilder.get())->Append(idx);
+  auto status = static_cast<arrow::Int32Builder*>(mBuilder.get())->Append(idx);
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot append to self-index array: %s", status.ToString().c_str());
+  }
 }
 
 std::shared_ptr<arrow::ChunkedArray> SelfBuilder::result() const
 {
   std::shared_ptr<arrow::Array> array;
   auto status = static_cast<arrow::Int32Builder*>(mBuilder.get())->Finish(&array);
   if (!status.ok()) {
-    cannotBuildAnArray();
+    cannotBuildAnArray(status.ToString().c_str());
   }
 
   return std::make_shared<arrow::ChunkedArray>(array);
@@ -93,7 +97,7 @@ SingleBuilder::SingleBuilder(std::shared_ptr<arrow::ChunkedArray> source, arrow:
 {
   auto status = arrow::MakeBuilder(pool, arrow::int32(), &mBuilder);
   if (!status.ok()) {
-    throw framework::runtime_error("Cannot create array builder for the single-valued index!");
+    throw framework::runtime_error_f("Cannot create array builder for the single-valued index: %s", status.ToString().c_str());
   }
 }
 
@@ -126,10 +130,14 @@ bool SingleBuilder::find(int idx)
 
 void SingleBuilder::fill(int idx)
 {
+  arrow::Status status;
   if (mPosition < mSourceSize && valueAt(mPosition) == idx) {
-    (void)static_cast<arrow::Int32Builder*>(mBuilder.get())->Append((int)mPosition);
+    status = static_cast<arrow::Int32Builder*>(mBuilder.get())->Append((int)mPosition);
   } else {
-    (void)static_cast<arrow::Int32Builder*>(mBuilder.get())->Append(-1);
+    status = static_cast<arrow::Int32Builder*>(mBuilder.get())->Append(-1);
+  }
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot append to array: %s", status.ToString().c_str());
   }
 }
 
@@ -138,22 +146,23 @@ std::shared_ptr<arrow::ChunkedArray> SingleBuilder::result() const
   std::shared_ptr<arrow::Array> array;
   auto status = static_cast<arrow::Int32Builder*>(mBuilder.get())->Finish(&array);
   if (!status.ok()) {
-    cannotBuildAnArray();
+    cannotBuildAnArray(status.ToString().c_str());
   }
   return std::make_shared<arrow::ChunkedArray>(array);
 }
 
 SliceBuilder::SliceBuilder(std::shared_ptr<arrow::ChunkedArray> source, arrow::MemoryPool* pool)
   : ChunkedArrayIterator{source}
 {
-  if (!preSlice().ok()) {
-    throw framework::runtime_error("Cannot pre-slice the source for slice-index building");
+  auto status = preSlice();
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot pre-slice the source for slice-index building: %s", status.ToString().c_str());
   }
 
   std::unique_ptr<arrow::ArrayBuilder> builder;
-  auto status = arrow::MakeBuilder(pool, arrow::int32(), &builder);
+  status = arrow::MakeBuilder(pool, arrow::int32(), &builder);
   if (!status.ok()) {
-    throw framework::runtime_error("Cannot create array for the slice-index builder!");
+    throw framework::runtime_error_f("Cannot create array for the slice-index builder: %s", status.ToString().c_str());
   }
   mListBuilder = std::make_unique<arrow::FixedSizeListBuilder>(pool, std::move(builder), 2);
   mValueBuilder = static_cast<arrow::FixedSizeListBuilder*>(mListBuilder.get())->value_builder();
@@ -166,8 +175,9 @@ void SliceBuilder::reset(std::shared_ptr<arrow::ChunkedArray> source)
   mListBuilder->Reset();
   mValuePos = 0;
   static_cast<ChunkedArrayIterator*>(this)->reset(source);
-  if (!preSlice().ok()) {
-    throw framework::runtime_error("Cannot pre-slice the source for slice-index building");
+  auto status = preSlice();
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot pre-slice the source for slice-index building: %s", status.ToString().c_str());
   }
 }
 
@@ -211,13 +221,17 @@ std::shared_ptr<arrow::ChunkedArray> SliceBuilder::result() const
   std::shared_ptr<arrow::Array> array;
   auto status = static_cast<arrow::FixedSizeListBuilder*>(mListBuilder.get())->Finish(&array);
   if (!status.ok()) {
-    cannotBuildAnArray();
+    cannotBuildAnArray(status.ToString().c_str());
   }
   return std::make_shared<arrow::ChunkedArray>(array);
 }
 
 arrow::Status SliceBuilder::SliceBuilder::preSlice()
 {
+  auto status = arrow::compute::Initialize();
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot initialize arrow compute: %s", status.ToString().c_str());
+  }
   arrow::Datum value_counts;
   auto options = arrow::compute::ScalarAggregateOptions::Defaults();
   ARROW_ASSIGN_OR_RAISE(value_counts, arrow::compute::CallFunction("value_counts", {mSource}, &options));
@@ -230,14 +244,15 @@ arrow::Status SliceBuilder::SliceBuilder::preSlice()
 ArrayBuilder::ArrayBuilder(std::shared_ptr<arrow::ChunkedArray> source, arrow::MemoryPool* pool)
   : ChunkedArrayIterator{source}
 {
-  if (!preFind().ok()) {
-    throw framework::runtime_error("Cannot pre-find in a source for array-index building");
+  auto&& status = preFind();
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot pre-find in a source for array-index building: %s", status.ToString().c_str());
   }
 
   std::unique_ptr<arrow::ArrayBuilder> builder;
-  auto status = arrow::MakeBuilder(pool, arrow::int32(), &builder);
+  status = arrow::MakeBuilder(pool, arrow::int32(), &builder);
   if (!status.ok()) {
-    throw framework::runtime_error("Cannot create array for the array-index builder!");
+    throw framework::runtime_error_f("Cannot create array for the array-index builder: %s", status.ToString().c_str());
   }
   mListBuilder = std::make_unique<arrow::ListBuilder>(pool, std::move(builder));
   mValueBuilder = static_cast<arrow::ListBuilder*>(mListBuilder.get())->value_builder();
@@ -246,8 +261,9 @@ ArrayBuilder::ArrayBuilder(std::shared_ptr<arrow::ChunkedArray> source, arrow::M
 void ArrayBuilder::reset(std::shared_ptr<arrow::ChunkedArray> source)
 {
   static_cast<ChunkedArrayIterator*>(this)->reset(source);
-  if (!preFind().ok()) {
-    throw framework::runtime_error("Cannot pre-find in a source for array-index building");
+  auto status = preFind();
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot pre-find in a source for array-index building: %s", status.ToString().c_str());
   }
   mValues.clear();
   mIndices.clear();
@@ -274,13 +290,17 @@ std::shared_ptr<arrow::ChunkedArray> ArrayBuilder::result() const
   std::shared_ptr<arrow::Array> array;
   auto status = static_cast<arrow::ListBuilder*>(mListBuilder.get())->Finish(&array);
   if (!status.ok()) {
-    cannotBuildAnArray();
+    cannotBuildAnArray(status.ToString().c_str());
   }
   return std::make_shared<arrow::ChunkedArray>(array);
 }
 
 arrow::Status ArrayBuilder::preFind()
 {
+  auto status = arrow::compute::Initialize();
+  if (!status.ok()) {
+    throw framework::runtime_error_f("Cannot initialize arrow compute: %s", status.ToString().c_str());
+  }
   arrow::Datum max;
   auto options = arrow::compute::ScalarAggregateOptions::Defaults();
   ARROW_ASSIGN_OR_RAISE(max, arrow::compute::CallFunction("max", {mSource}, &options));