Fixing 375 (#410)

* partially fixing #375

* adding __len__ to df

* passing pycylon.Table to df operations

* fixing buffer copy bug

* logging status errors in macros

* bug fix

* test case for chunked tables

* cosmetic changes

* fix #375

* fix #375

cpp/src/cylon/arrow/arrow_kernels.cpp

@@ -316,11 +316,22 @@ class NumericInplaceIndexSortKernel : public InplaceIndexSortKernel {
     auto array = std::static_pointer_cast<arrow::NumericArray<TYPE>>(values);
     const std::shared_ptr<arrow::ArrayData> &data = array->data();
     // get the first buffer as a mutable buffer
-    T *left_data = data->GetMutableValues<T>(1);
+    if (!util::IsMutable(values)) {
+      return arrow::Status::ExecutionError("inplace sort called on an array with immutable buffers");
+    }
+
+    T *left_data;
+    // todo: this is a temp fix for arrow bug #ARROW-12495 (https://issues.apache.org/jira/browse/ARROW-12495)
+    // todo: remove this once the #ARROW-1245 is fixed
+    if (data->buffers[1]->mutable_data() == nullptr) {
+      left_data = reinterpret_cast<T *>(const_cast<uint8_t *>(data->buffers[1]->data()));
+    } else {
+      left_data = data->template GetMutableValues<T>(1);
+    }
     int64_t length = values->length();
     int64_t buf_size = length * sizeof(uint64_t);
 
-    arrow::Result<std::unique_ptr<arrow::Buffer>> result = AllocateBuffer(buf_size + 1, pool_);
+    arrow::Result<std::unique_ptr<arrow::Buffer>> result = AllocateBuffer(buf_size, pool_);
     RETURN_ARROW_STATUS_IF_FAILED(result.status());
     std::shared_ptr<arrow::Buffer> indices_buf(std::move(result.ValueOrDie()));
 

cpp/src/cylon/util/arrow_utils.hpp

@@ -15,7 +15,7 @@
 #ifndef CYLON_SRC_UTIL_ARROW_UTILS_HPP_
 #define CYLON_SRC_UTIL_ARROW_UTILS_HPP_
 
-#include <arrow/compute/kernel.h>
+#include <arrow/api.h>
 #include <arrow/table.h>
 
 namespace cylon {
@@ -122,6 +122,13 @@ arrow::Status SampleArray(const std::shared_ptr<arrow::Array> &array,
 
 std::shared_ptr<arrow::Array> GetChunkOrEmptyArray(const std::shared_ptr<arrow::ChunkedArray> &column, int chunk);
 
+inline bool IsMutable(const std::shared_ptr<arrow::Array> &array) {
+  for (auto &&buff: array->data()->buffers) {
+    if (buff != nullptr && !buff->is_mutable()) return false;
+  }
+  return true;
+}
+
 }  // namespace util
 }  // namespace cylon
 #endif  // CYLON_SRC_UTIL_ARROW_UTILS_HPP_

python/test/test_pd_read_csv.py

@@ -0,0 +1,40 @@
+##
+# Licensed 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.
+##
+
+'''
+Run test:
+>> pytest -q python/test/test_pd_read_csv.py
+'''
+
+import pandas as pd
+
+from pycylon import DataFrame, CylonEnv
+from pycylon.net import MPIConfig
+
+
+def test_pd_read_csv():
+    env = CylonEnv(config=MPIConfig())
+
+    df1 = DataFrame(pd.read_csv('/tmp/user_usage_tm_1.csv'))
+    df2 = DataFrame(pd.read_csv('/tmp/user_device_tm_1.csv'))
+
+    df1 = df1.set_index([3], drop=True)
+    df2 = df2.set_index([0], drop=True)
+
+    df1.to_table().retain_memory(False)
+    df2.to_table().retain_memory(False)
+
+    df3 = df1.merge(right=df2, left_on=[3], right_on=[0], algorithm='sort', env=env)
+
+    assert len(df3)