ARROW-3459: [C++][Gandiva] support for string o/p

- If the output vectors aren't provided, allow resizable
  data buffers.
- If the output vectors are provided, assert that the data
  buffer is resizeable.
- use a cpp function to write to string-like o/p buffers,
  this checks for capacity and updates the offset vector.

Author: Pindikura Ravindra

cpp/src/gandiva/annotator.cc

@@ -31,30 +31,35 @@ FieldDescriptorPtr Annotator::CheckAndAddInputFieldDescriptor(FieldPtr field) {
     return found->second;
   }
 
-  auto desc = MakeDesc(field);
+  auto desc = MakeDesc(field, false /*is_output*/);
   in_name_to_desc_[field->name()] = desc;
   return desc;
 }
 
 FieldDescriptorPtr Annotator::AddOutputFieldDescriptor(FieldPtr field) {
-  auto desc = MakeDesc(field);
+  auto desc = MakeDesc(field, true /*is_output*/);
   out_descs_.push_back(desc);
   return desc;
 }
 
-FieldDescriptorPtr Annotator::MakeDesc(FieldPtr field) {
+FieldDescriptorPtr Annotator::MakeDesc(FieldPtr field, bool is_output) {
   int data_idx = buffer_count_++;
   int validity_idx = buffer_count_++;
   int offsets_idx = FieldDescriptor::kInvalidIdx;
   if (arrow::is_binary_like(field->type()->id())) {
     offsets_idx = buffer_count_++;
   }
-  return std::make_shared<FieldDescriptor>(field, data_idx, validity_idx, offsets_idx);
+  int data_buffer_ptr_idx = FieldDescriptor::kInvalidIdx;
+  if (is_output) {
+    data_buffer_ptr_idx = buffer_count_++;
+  }
+  return std::make_shared<FieldDescriptor>(field, data_idx, validity_idx, offsets_idx,
+                                           data_buffer_ptr_idx);
 }
 
 void Annotator::PrepareBuffersForField(const FieldDescriptor& desc,
                                        const arrow::ArrayData& array_data,
-                                       EvalBatch* eval_batch) {
+                                       EvalBatch* eval_batch, bool is_output) {
   int buffer_idx = 0;
 
   // The validity buffer is optional. Use nullptr if it does not have one.
@@ -74,7 +79,12 @@ void Annotator::PrepareBuffersForField(const FieldDescriptor& desc,
 
   uint8_t* data_buf = const_cast<uint8_t*>(array_data.buffers[buffer_idx]->data());
   eval_batch->SetBuffer(desc.data_idx(), data_buf);
-  ++buffer_idx;
+  if (is_output) {
+    // pass in the Buffer object for output data buffers. Can be used for resizing.
+    uint8_t* data_buf_ptr =
+        reinterpret_cast<uint8_t*>(array_data.buffers[buffer_idx].get());
+    eval_batch->SetBuffer(desc.data_buffer_ptr_idx(), data_buf_ptr);
+  }
 }
 
 EvalBatchPtr Annotator::PrepareEvalBatch(const arrow::RecordBatch& record_batch,
@@ -92,14 +102,14 @@ EvalBatchPtr Annotator::PrepareEvalBatch(const arrow::RecordBatch& record_batch,
     }
 
     PrepareBuffersForField(*(found->second), *(record_batch.column(i))->data(),
-                           eval_batch.get());
+                           eval_batch.get(), false /*is_output*/);
   }
 
   // Fill in the entries for the output fields.
   int idx = 0;
   for (auto& arraydata : out_vector) {
     const FieldDescriptorPtr& desc = out_descs_.at(idx);
-    PrepareBuffersForField(*desc, *arraydata, eval_batch.get());
+    PrepareBuffersForField(*desc, *arraydata, eval_batch.get(), true /*is_output*/);
     ++idx;
   }
   return eval_batch;

cpp/src/gandiva/annotator.h

@@ -54,12 +54,13 @@ class GANDIVA_EXPORT Annotator {
 
  private:
   /// Annotate a field and return the descriptor.
-  FieldDescriptorPtr MakeDesc(FieldPtr field);
+  FieldDescriptorPtr MakeDesc(FieldPtr field, bool is_output);
 
   /// Populate eval_batch by extracting the raw buffers from the arrow array, whose
   /// contents are represent by the annotated descriptor 'desc'.
   void PrepareBuffersForField(const FieldDescriptor& desc,
-                              const arrow::ArrayData& array_data, EvalBatch* eval_batch);
+                              const arrow::ArrayData& array_data, EvalBatch* eval_batch,
+                              bool is_output);
 
   /// The list of input/output buffers (includes bitmap buffers, value buffers and
   /// offset buffers).

cpp/src/gandiva/annotator_test.cc

@@ -73,6 +73,7 @@ TEST_F(TestAnnotator, TestAdd) {
   EXPECT_EQ(desc_sum->field(), field_sum);
   EXPECT_EQ(desc_sum->data_idx(), 4);
   EXPECT_EQ(desc_sum->validity_idx(), 5);
+  EXPECT_EQ(desc_sum->data_buffer_ptr_idx(), 6);
 
   // prepare record batch
   int num_records = 100;
@@ -85,7 +86,7 @@ TEST_F(TestAnnotator, TestAdd) {
 
   auto arrow_sum = MakeInt32Array(num_records);
   EvalBatchPtr batch = annotator.PrepareEvalBatch(*record_batch, {arrow_sum->data()});
-  EXPECT_EQ(batch->GetNumBuffers(), 6);
+  EXPECT_EQ(batch->GetNumBuffers(), 7);
 
   auto buffers = batch->GetBufferArray();
   EXPECT_EQ(buffers[desc_a->validity_idx()], arrow_v0->data()->buffers.at(0)->data());
@@ -94,6 +95,8 @@ TEST_F(TestAnnotator, TestAdd) {
   EXPECT_EQ(buffers[desc_b->data_idx()], arrow_v1->data()->buffers.at(1)->data());
   EXPECT_EQ(buffers[desc_sum->validity_idx()], arrow_sum->data()->buffers.at(0)->data());
   EXPECT_EQ(buffers[desc_sum->data_idx()], arrow_sum->data()->buffers.at(1)->data());
+  EXPECT_EQ(buffers[desc_sum->data_buffer_ptr_idx()],
+            reinterpret_cast<uint8_t*>(arrow_sum->data()->buffers.at(1).get()));
 
   auto bitmaps = batch->GetLocalBitMapArray();
   EXPECT_EQ(bitmaps, nullptr);

cpp/src/gandiva/expr_validator.cc

@@ -89,6 +89,12 @@ Status ExprValidator::Visit(const IfNode& node) {
   auto then_node_ret_type = node.then_node()->return_type();
   auto else_node_ret_type = node.else_node()->return_type();
 
+  // condition must be of boolean type.
+  ARROW_RETURN_IF(
+      !node.condition()->return_type()->Equals(arrow::boolean()),
+      Status::ExpressionValidationError("condition must be of boolean type, found type ",
+                                        node.condition()->return_type()->ToString()));
+
   // Then-branch return type must match.
   ARROW_RETURN_IF(!if_node_ret_type->Equals(*then_node_ret_type),
                   Status::ExpressionValidationError(

cpp/src/gandiva/field_descriptor.h

@@ -31,11 +31,12 @@ class FieldDescriptor {
   static const int kInvalidIdx = -1;
 
   FieldDescriptor(FieldPtr field, int data_idx, int validity_idx = kInvalidIdx,
-                  int offsets_idx = kInvalidIdx)
+                  int offsets_idx = kInvalidIdx, int data_buffer_ptr_idx = kInvalidIdx)
       : field_(field),
         data_idx_(data_idx),
         validity_idx_(validity_idx),
-        offsets_idx_(offsets_idx) {}
+        offsets_idx_(offsets_idx),
+        data_buffer_ptr_idx_(data_buffer_ptr_idx) {}
 
   /// Index of validity array in the array-of-buffers
   int validity_idx() const { return validity_idx_; }
@@ -46,18 +47,24 @@ class FieldDescriptor {
   /// Index of offsets array in the array-of-buffers
   int offsets_idx() const { return offsets_idx_; }
 
+  /// Index of data buffer pointer in the array-of-buffers
+  int data_buffer_ptr_idx() const { return data_buffer_ptr_idx_; }
+
   FieldPtr field() const { return field_; }
 
   const std::string& Name() const { return field_->name(); }
   DataTypePtr Type() const { return field_->type(); }
 
   bool HasOffsetsIdx() const { return offsets_idx_ != kInvalidIdx; }
 
+  bool HasDataBufferPtrIdx() const { return data_buffer_ptr_idx_ != kInvalidIdx; }
+
  private:
   FieldPtr field_;
   int data_idx_;
   int validity_idx_;
   int offsets_idx_;
+  int data_buffer_ptr_idx_;
 };
 
 }  // namespace gandiva

cpp/src/gandiva/gdv_function_stubs.cc

@@ -72,6 +72,31 @@ bool gdv_fn_in_expr_lookup_utf8(int64_t ptr, const char* data, int data_len,
       reinterpret_cast<gandiva::InHolder<std::string>*>(ptr);
   return holder->HasValue(std::string(data, data_len));
 }
+
+int32_t gdv_fn_populate_varlen_vector(int64_t context_ptr, int8_t* data_ptr,
+                                      int32_t* offsets, int64_t slot,
+                                      const char* entry_buf, int32_t entry_len) {
+  auto buffer = reinterpret_cast<arrow::ResizableBuffer*>(data_ptr);
+  int32_t offset = static_cast<int32_t>(buffer->size());
+
+  // This also sets the size in the buffer.
+  auto status = buffer->Resize(offset + entry_len, false /*shrink*/);
+  if (!status.ok()) {
+    gandiva::ExecutionContext* context =
+        reinterpret_cast<gandiva::ExecutionContext*>(context_ptr);
+
+    context->set_error_msg(status.message().c_str());
+    return -1;
+  }
+
+  // append the new entry.
+  memcpy(buffer->mutable_data() + offset, entry_buf, entry_len);
+
+  // update offsets buffer.
+  offsets[slot] = offset;
+  offsets[slot + 1] = offset + entry_len;
+  return 0;
+}
 }
 
 namespace gandiva {
@@ -135,6 +160,18 @@ void ExportedStubFunctions::AddMappings(Engine* engine) const {
   engine->AddGlobalMappingForFunc("gdv_fn_in_expr_lookup_utf8",
                                   types->i1_type() /*return_type*/, args,
                                   reinterpret_cast<void*>(gdv_fn_in_expr_lookup_utf8));
+
+  // gdv_fn_populate_varlen_vector
+  args = {types->i64_type(),      // int64_t execution_context
+          types->i8_ptr_type(),   // int8_t* data ptr
+          types->i32_ptr_type(),  // int32_t* offsets ptr
+          types->i64_type(),      // int64_t slot
+          types->i8_ptr_type(),   // const char* entry_buf
+          types->i32_type()};     // int32_t entry__len
+
+  engine->AddGlobalMappingForFunc("gdv_fn_populate_varlen_vector",
+                                  types->i32_type() /*return_type*/, args,
+                                  reinterpret_cast<void*>(gdv_fn_populate_varlen_vector));
 }
 
 }  // namespace gandiva

cpp/src/gandiva/jni/jni_common.cc

@@ -632,6 +632,32 @@ JNIEXPORT jlong JNICALL Java_org_apache_arrow_gandiva_evaluator_JniWrapper_build
   return module_id;
 }
 
+///
+/// \brief Resizable buffer which resizes by doing a callback into java.
+///
+class JavaResizableBuffer : public arrow::ResizableBuffer {
+ public:
+  JavaResizableBuffer(uint8_t* buffer, int32_t len) : ResizableBuffer(buffer, len) {
+    size_ = 0;
+  }
+
+  Status Resize(const int64_t new_size, bool shrink_to_fit) override {
+    if (shrink_to_fit == true) {
+      return Status::NotImplemented("shrink not implemented");
+    } else if (new_size < capacity()) {
+      size_ = new_size;
+      return Status::OK();
+    } else {
+      // TODO: callback into java to re-alloc the buffer.
+      return Status::NotImplemented("buffer expand not implemented");
+    }
+  }
+
+  Status Reserve(const int64_t new_capacity) override {
+    return Status::NotImplemented("reserve not implemented");
+  }
+};
+
 #define CHECK_OUT_BUFFER_IDX_AND_BREAK(idx, len)                               \
   if (idx >= len) {                                                            \
     status = gandiva::Status::Invalid("insufficient number of out_buf_addrs"); \
@@ -710,20 +736,31 @@ Java_org_apache_arrow_gandiva_evaluator_JniWrapper_evaluateProjector(
     int buf_idx = 0;
     int sz_idx = 0;
     for (FieldPtr field : ret_types) {
+      std::vector<std::shared_ptr<arrow::Buffer>> buffers;
+
       CHECK_OUT_BUFFER_IDX_AND_BREAK(buf_idx, out_bufs_len);
       uint8_t* validity_buf = reinterpret_cast<uint8_t*>(out_bufs[buf_idx++]);
       jlong bitmap_sz = out_sizes[sz_idx++];
-      std::shared_ptr<arrow::MutableBuffer> bitmap_buf =
-          std::make_shared<arrow::MutableBuffer>(validity_buf, bitmap_sz);
+      buffers.push_back(std::make_shared<arrow::MutableBuffer>(validity_buf, bitmap_sz));
+
+      if (arrow::is_binary_like(field->type()->id())) {
+        CHECK_OUT_BUFFER_IDX_AND_BREAK(buf_idx, out_bufs_len);
+        uint8_t* offsets_buf = reinterpret_cast<uint8_t*>(out_bufs[buf_idx++]);
+        jlong offsets_sz = out_sizes[sz_idx++];
+        buffers.push_back(
+            std::make_shared<arrow::MutableBuffer>(offsets_buf, offsets_sz));
+      }
 
       CHECK_OUT_BUFFER_IDX_AND_BREAK(buf_idx, out_bufs_len);
       uint8_t* value_buf = reinterpret_cast<uint8_t*>(out_bufs[buf_idx++]);
       jlong data_sz = out_sizes[sz_idx++];
-      std::shared_ptr<arrow::MutableBuffer> data_buf =
-          std::make_shared<arrow::MutableBuffer>(value_buf, data_sz);
+      if (arrow::is_binary_like(field->type()->id())) {
+        buffers.push_back(std::make_shared<JavaResizableBuffer>(value_buf, data_sz));
+      } else {
+        buffers.push_back(std::make_shared<arrow::MutableBuffer>(value_buf, data_sz));
+      }
 
-      auto array_data =
-          arrow::ArrayData::Make(field->type(), output_row_count, {bitmap_buf, data_buf});
+      auto array_data = arrow::ArrayData::Make(field->type(), output_row_count, buffers);
       output.push_back(array_data);
     }
     status = holder->projector()->Evaluate(*in_batch, selection_vector.get(), output);

cpp/src/gandiva/llvm_generator.cc

@@ -155,6 +155,14 @@ llvm::Value* LLVMGenerator::GetValidityReference(llvm::Value* arg_addrs, int idx
   return ir_builder()->CreateIntToPtr(load, types()->i64_ptr_type(), name + "_varray");
 }
 
+/// Get reference to data array at specified index in the args list.
+llvm::Value* LLVMGenerator::GetDataBufferPtrReference(llvm::Value* arg_addrs, int idx,
+                                                      FieldPtr field) {
+  const std::string& name = field->name();
+  llvm::Value* load = LoadVectorAtIndex(arg_addrs, idx, name);
+  return ir_builder()->CreateIntToPtr(load, types()->i8_ptr_type(), name + "_buf_ptr");
+}
+
 /// Get reference to data array at specified index in the args list.
 llvm::Value* LLVMGenerator::GetDataReference(llvm::Value* arg_addrs, int idx,
                                              FieldPtr field) {
@@ -293,6 +301,10 @@ Status LLVMGenerator::CodeGenExprValue(DexPtr value_expr, FieldDescriptorPtr out
   builder->SetInsertPoint(loop_entry);
   llvm::Value* output_ref =
       GetDataReference(arg_addrs, output->data_idx(), output->field());
+  llvm::Value* output_buffer_ptr_ref = GetDataBufferPtrReference(
+      arg_addrs, output->data_buffer_ptr_idx(), output->field());
+  llvm::Value* output_offset_ref =
+      GetOffsetsReference(arg_addrs, output->offsets_idx(), output->field());
 
   // Loop body
   builder->SetInsertPoint(loop_body);
@@ -323,13 +335,21 @@ Status LLVMGenerator::CodeGenExprValue(DexPtr value_expr, FieldDescriptorPtr out
 
   // save the value in the output vector.
   builder->SetInsertPoint(loop_body_tail);
+
   auto output_type_id = output->Type()->id();
   if (output_type_id == arrow::Type::BOOL) {
     SetPackedBitValue(output_ref, loop_var, output_value->data());
   } else if (arrow::is_primitive(output_type_id) ||
              output_type_id == arrow::Type::DECIMAL) {
     llvm::Value* slot_offset = builder->CreateGEP(output_ref, loop_var);
     builder->CreateStore(output_value->data(), slot_offset);
+  } else if (arrow::is_binary_like(output_type_id)) {
+    // Var-len output. Make a function call to populate the data.
+    // if there is an error, the fn sets it in the context. And, will be returned at the
+    // end of this row batch.
+    AddFunctionCall("gdv_fn_populate_varlen_vector", types()->i32_type(),
+                    {arg_context_ptr, output_buffer_ptr_ref, output_offset_ref, loop_var,
+                     output_value->data(), output_value->length()});
   } else {
     return Status::NotImplemented("output type ", output->Type()->ToString(),
                                   " not supported");

cpp/src/gandiva/llvm_generator.h

@@ -180,6 +180,9 @@ class GANDIVA_EXPORT LLVMGenerator {
   /// Generate code to load the vector at specified index and cast it as offsets array.
   llvm::Value* GetOffsetsReference(llvm::Value* arg_addrs, int idx, FieldPtr field);
 
+  /// Generate code to load the vector at specified index and cast it as buffer pointer.
+  llvm::Value* GetDataBufferPtrReference(llvm::Value* arg_addrs, int idx, FieldPtr field);
+
   /// Generate code for the value array of one expression.
   Status CodeGenExprValue(DexPtr value_expr, FieldDescriptorPtr output, int suffix_idx,
                           llvm::Function** fn,