merge

c_glib/plasma-glib/client.cpp

@@ -185,6 +185,7 @@ gplasma_client_create_options_get_metadata(GPlasmaClientCreateOptions *options,
 
 typedef struct GPlasmaClientPrivate_ {
   plasma::PlasmaClient *client;
+  bool disconnected;
 } GPlasmaClientPrivate;
 
 enum {
@@ -205,10 +206,12 @@ gplasma_client_finalize(GObject *object)
 {
   auto priv = GPLASMA_CLIENT_GET_PRIVATE(object);
 
-  auto status = priv->client->Disconnect();
-  if (!status.ok()) {
-    g_warning("[plasma][client][finalize] Failed to disconnect: %s",
-              status.ToString().c_str());
+  if (!priv->disconnected) {
+    auto status = priv->client->Disconnect();
+    if (!status.ok()) {
+      g_warning("[plasma][client][finalize] Failed to disconnect: %s",
+                status.ToString().c_str());
+    }
   }
   delete priv->client;
 
@@ -431,6 +434,29 @@ gplasma_client_refer_object(GPlasmaClient *client,
   }
 }
 
+/**
+ * gplasma_client_disconnect:
+ * @client: A #GPlasmaClient.
+ * @error: (nullable): Return location for a #GError or %NULL.
+ *
+ * Returns: %TRUE on success, %FALSE if there was an error.
+ *
+ * Since: 0.12.0
+ */
+gboolean
+gplasma_client_disconnect(GPlasmaClient *client,
+                          GError **error)
+{
+  auto priv = GPLASMA_CLIENT_GET_PRIVATE(client);
+  auto status = priv->client->Disconnect();
+  if (garrow_error_check(error, status, "[plasma][client][disconnect]")) {
+    priv->disconnected = true;
+    return TRUE;
+  } else {
+    return FALSE;
+  }
+}
+
 G_END_DECLS
 
 GPlasmaClient *

c_glib/plasma-glib/client.h

@@ -71,5 +71,7 @@ gplasma_client_refer_object(GPlasmaClient *client,
                             GPlasmaObjectID *id,
                             gint64 timeout_ms,
                             GError **error);
+gboolean gplasma_client_disconnect(GPlasmaClient *client,
+                                   GError **error);
 
 G_END_DECLS

c_glib/test/plasma/test-plasma-client.rb

@@ -24,6 +24,9 @@ class TestPlasmaClient < Test::Unit::TestCase
     @store = Helper::PlasmaStore.new
     @store.start
     @client = Plasma::Client.new(@store.socket_path)
+    @id = Plasma::ObjectID.new("Hello")
+    @data = "World"
+    @options = Plasma::ClientCreateOptions.new
   end
 
   def teardown
@@ -34,10 +37,7 @@ class TestPlasmaClient < Test::Unit::TestCase
     def setup
       super
 
-      @id = Plasma::ObjectID.new("Hello")
-      @data = "World"
       @metadata = "Metadata"
-      @options = Plasma::ClientCreateOptions.new
     end
 
     test("no options") do
@@ -84,4 +84,11 @@ class TestPlasmaClient < Test::Unit::TestCase
                    ])
     end
   end
+
+  test("#disconnect") do
+    @client.disconnect
+    assert_raise(Arrow::Error::Io) do
+      @client.create(@id, @data.bytesize, @options)
+    end
+  end
 end

cpp/src/arrow/array.h

@@ -87,7 +87,7 @@ class Status;
 /// input array and replace them with newly-allocated data, changing the output
 /// data type as well.
 struct ARROW_EXPORT ArrayData {
-  ArrayData() : length(0) {}
+  ArrayData() : length(0), null_count(0), offset(0) {}
 
   ArrayData(const std::shared_ptr<DataType>& type, int64_t length,
             int64_t null_count = kUnknownNullCount, int64_t offset = 0)
@@ -311,7 +311,7 @@ class ARROW_EXPORT Array {
   std::string ToString() const;
 
  protected:
-  Array() {}
+  Array() : null_bitmap_data_(NULLPTR) {}
 
   std::shared_ptr<ArrayData> data_;
   const uint8_t* null_bitmap_data_;
@@ -382,7 +382,7 @@ class ARROW_EXPORT PrimitiveArray : public FlatArray {
   std::shared_ptr<Buffer> values() const { return data_->buffers[1]; }
 
  protected:
-  PrimitiveArray() {}
+  PrimitiveArray() : raw_values_(NULLPTR) {}
 
   inline void SetData(const std::shared_ptr<ArrayData>& data) {
     auto values = data->buffers[1];
@@ -565,7 +565,7 @@ class ARROW_EXPORT BinaryArray : public FlatArray {
 
  protected:
   // For subclasses
-  BinaryArray() {}
+  BinaryArray() : raw_value_offsets_(NULLPTR), raw_data_(NULLPTR) {}
 
   /// Protected method for constructors
   void SetData(const std::shared_ptr<ArrayData>& data);

cpp/src/arrow/compute/kernel.h

@@ -19,6 +19,7 @@
 #define ARROW_COMPUTE_KERNEL_H
 
 #include <memory>
+#include <utility>
 #include <vector>
 
 #include "arrow/array.h"
@@ -78,6 +79,14 @@ struct ARROW_EXPORT Datum {
 
   Datum(const Datum& other) noexcept { this->value = other.value; }
 
+  // Define move constructor and move assignment, for better performance
+  Datum(Datum&& other) noexcept : value(std::move(other.value)) {}
+
+  Datum& operator=(Datum&& other) noexcept {
+    value = std::move(other.value);
+    return *this;
+  }
+
   Datum::type kind() const {
     switch (this->value.which()) {
       case 0:

cpp/src/arrow/io/buffered-test.cc

@@ -67,7 +67,7 @@ class FileTestFixture : public ::testing::Test {
 
   void EnsureFileDeleted() {
     if (FileExists(path_)) {
-      std::remove(path_.c_str());
+      ARROW_UNUSED(std::remove(path_.c_str()));
     }
   }
 

cpp/src/arrow/io/file-test.cc

@@ -56,7 +56,7 @@ class FileTestFixture : public ::testing::Test {
 
   void EnsureFileDeleted() {
     if (FileExists(path_)) {
-      std::remove(path_.c_str());
+      ARROW_UNUSED(std::remove(path_.c_str()));
     }
   }
 

cpp/src/arrow/io/test-common.h

@@ -118,7 +118,7 @@ class MemoryMapFixture {
  public:
   void TearDown() {
     for (auto path : tmp_files_) {
-      std::remove(path.c_str());
+      ARROW_UNUSED(std::remove(path.c_str()));
     }
   }
 

cpp/src/arrow/ipc/json-integration-test.cc

@@ -262,7 +262,7 @@ class TestJSONIntegration : public ::testing::Test {
 
   void TearDown() {
     for (const std::string path : tmp_paths_) {
-      std::remove(path.c_str());
+      ARROW_UNUSED(std::remove(path.c_str()));
     }
   }
 

cpp/src/arrow/ipc/json.cc

@@ -99,7 +99,7 @@ Status JsonWriter::WriteRecordBatch(const RecordBatch& batch) {
 class JsonReader::JsonReaderImpl {
  public:
   JsonReaderImpl(MemoryPool* pool, const std::shared_ptr<Buffer>& data)
-      : pool_(pool), data_(data) {}
+      : pool_(pool), data_(data), record_batches_(nullptr) {}
 
   Status ParseAndReadSchema() {
     doc_.Parse(reinterpret_cast<const rj::Document::Ch*>(data_->data()),

cpp/src/arrow/ipc/writer.cc

@@ -772,7 +772,10 @@ class SchemaWriter : public StreamBookKeeper {
  public:
   SchemaWriter(const Schema& schema, DictionaryMemo* dictionary_memo, MemoryPool* pool,
                io::OutputStream* sink)
-      : StreamBookKeeper(sink), schema_(schema), dictionary_memo_(dictionary_memo) {}
+      : StreamBookKeeper(sink),
+        pool_(pool),
+        schema_(schema),
+        dictionary_memo_(dictionary_memo) {}
 
   Status WriteSchema() {
 #ifndef NDEBUG

cpp/src/arrow/table.cc

@@ -392,7 +392,7 @@ class SimpleTable : public Table {
   std::vector<std::shared_ptr<Column>> columns_;
 };
 
-Table::Table() {}
+Table::Table() : num_rows_(0) {}
 
 std::shared_ptr<Table> Table::Make(const std::shared_ptr<Schema>& schema,
                                    const std::vector<std::shared_ptr<Column>>& columns,

cpp/src/arrow/util/decimal.cc

@@ -889,7 +889,7 @@ Status Decimal128::Rescale(int32_t original_scale, int32_t new_scale,
 }
 
 // Helper function used by Decimal128::FromBigEndian
-static inline uint64_t FromBigEndian(const uint8_t* bytes, int32_t length) {
+static inline uint64_t UInt64FromBigEndian(const uint8_t* bytes, int32_t length) {
   // We don't bounds check the length here because this is called by
   // FromBigEndian that has a Decimal128 as its out parameters and
   // that function is already checking the length of the bytes and only
@@ -906,8 +906,7 @@ Status Decimal128::FromBigEndian(const uint8_t* bytes, int32_t length, Decimal12
   static constexpr int32_t kMinDecimalBytes = 1;
   static constexpr int32_t kMaxDecimalBytes = 16;
 
-  int64_t high;
-  uint64_t low;
+  int64_t high, low;
 
   if (length < kMinDecimalBytes || length > kMaxDecimalBytes) {
     std::ostringstream stream;
@@ -917,35 +916,45 @@ Status Decimal128::FromBigEndian(const uint8_t* bytes, int32_t length, Decimal12
     return Status::Invalid(stream.str());
   }
 
-  /// Bytes are coming in big-endian, so the first byte is the MSB and therefore holds the
-  /// sign bit.
+  // Bytes are coming in big-endian, so the first byte is the MSB and therefore holds the
+  // sign bit.
   const bool is_negative = static_cast<int8_t>(bytes[0]) < 0;
 
-  /// Sign extend the low bits if necessary
-  low = UINT64_MAX * (is_negative && length < 8);
-  high = -1 * (is_negative && length < kMaxDecimalBytes);
-
-  /// Stop byte of the high bytes
+  // 1. Extract the high bytes
+  // Stop byte of the high bytes
   const int32_t high_bits_offset = std::max(0, length - 8);
+  const auto high_bits = UInt64FromBigEndian(bytes, high_bits_offset);
 
-  /// Shift left enough bits to make room for the incoming int64_t
-  high <<= high_bits_offset * CHAR_BIT;
+  if (high_bits_offset == 8) {
+    // Avoid undefined shift by 64 below
+    high = high_bits;
+  } else {
+    high = -1 * (is_negative && length < kMaxDecimalBytes);
+    // Shift left enough bits to make room for the incoming int64_t
+    high <<= high_bits_offset * CHAR_BIT;
+    // Preserve the upper bits by inplace OR-ing the int64_t
+    high |= high_bits;
+  }
 
-  /// Preserve the upper bits by inplace OR-ing the int64_t
-  uint64_t value = arrow::FromBigEndian(bytes, high_bits_offset);
-  high |= value;
-
-  /// Stop byte of the low bytes
+  // 2. Extract the low bytes
+  // Stop byte of the low bytes
   const int32_t low_bits_offset = std::min(length, 8);
+  const auto low_bits =
+      UInt64FromBigEndian(bytes + high_bits_offset, length - high_bits_offset);
 
-  /// Shift left enough bits to make room for the incoming uint64_t
-  low <<= low_bits_offset * CHAR_BIT;
+  if (low_bits_offset == 8) {
+    // Avoid undefined shift by 64 below
+    low = low_bits;
+  } else {
+    // Sign extend the low bits if necessary
+    low = -1 * (is_negative && length < 8);
+    // Shift left enough bits to make room for the incoming int64_t
+    low <<= low_bits_offset * CHAR_BIT;
+    // Preserve the upper bits by inplace OR-ing the int64_t
+    low |= low_bits;
+  }
 
-  /// Preserve the upper bits by inplace OR-ing the uint64_t
-  value = arrow::FromBigEndian(bytes + high_bits_offset, length - high_bits_offset);
-  low |= value;
-
-  *out = Decimal128(high, low);
+  *out = Decimal128(high, static_cast<uint64_t>(low));
   return Status::OK();
 }
 

cpp/src/arrow/util/logging.h

@@ -22,6 +22,7 @@
 #include <memory>
 #include <string>
 
+#include "arrow/util/macros.h"
 #include "arrow/util/visibility.h"
 
 namespace arrow {
@@ -155,6 +156,8 @@ class ARROW_EXPORT ArrowLog : public ArrowLogBase {
   static void InstallFailureSignalHandler();
 
  private:
+  ARROW_DISALLOW_COPY_AND_ASSIGN(ArrowLog);
+
   // Hide the implementation of log provider by void *.
   // Otherwise, lib user may define the same macro to use the correct header file.
   void* logging_provider_;

cpp/src/arrow/util/rle-encoding-test.cc

@@ -193,7 +193,7 @@ void ValidateRle(const vector<int>& values, int bit_width, uint8_t* expected_enc
     EXPECT_EQ(encoded_len, expected_len);
   }
   if (expected_encoding != NULL) {
-    EXPECT_EQ(memcmp(buffer, expected_encoding, expected_len), 0);
+    EXPECT_EQ(memcmp(buffer, expected_encoding, encoded_len), 0);
   }
 
   // Verify read

cpp/src/arrow/util/rle-encoding.h

@@ -436,6 +436,7 @@ bool RleDecoder::NextCounts() {
     literal_count_ = (indicator_value >> 1) * 8;
   } else {
     repeat_count_ = indicator_value >> 1;
+    // XXX (ARROW-4018) this is not big-endian compatible
     bool result =
         bit_reader_.GetAligned<T>(static_cast<int>(BitUtil::CeilDiv(bit_width_, 8)),
                                   reinterpret_cast<T*>(&current_value_));

cpp/src/parquet/metadata.cc

@@ -47,6 +47,18 @@ const ApplicationVersion& ApplicationVersion::PARQUET_CPP_FIXED_STATS_VERSION()
   return version;
 }
 
+std::string ParquetVersionToString(ParquetVersion::type ver) {
+  switch (ver) {
+    case ParquetVersion::PARQUET_1_0:
+      return "1.0";
+    case ParquetVersion::PARQUET_2_0:
+      return "2.0";
+  }
+
+  // This should be unreachable
+  return "UNKNOWN";
+}
+
 template <typename DType>
 static std::shared_ptr<RowGroupStatistics> MakeTypedColumnStats(
     const format::ColumnMetaData& metadata, const ColumnDescriptor* descr) {

cpp/src/parquet/metadata.h

@@ -282,6 +282,8 @@ class PARQUET_EXPORT FileMetaDataBuilder {
   std::unique_ptr<FileMetaDataBuilderImpl> impl_;
 };
 
+PARQUET_EXPORT std::string ParquetVersionToString(ParquetVersion::type ver);
+
 }  // namespace parquet
 
 #endif  // PARQUET_FILE_METADATA_H

cpp/src/parquet/printer.cc

@@ -38,7 +38,7 @@ void ParquetFilePrinter::DebugPrint(std::ostream& stream, std::list<int> selecte
   const FileMetaData* file_metadata = fileReader->metadata().get();
 
   stream << "File Name: " << filename << "\n";
-  stream << "Version: " << file_metadata->version() << "\n";
+  stream << "Version: " << ParquetVersionToString(file_metadata->version()) << "\n";
   stream << "Created By: " << file_metadata->created_by() << "\n";
   stream << "Total rows: " << file_metadata->num_rows() << "\n";
 

cpp/src/plasma/client.cc

@@ -83,9 +83,6 @@ typedef struct XXH64_state_s XXH64_state_t;
 constexpr int64_t kHashingConcurrency = 8;
 constexpr int64_t kBytesInMB = 1 << 20;
 
-// Use 100MB as an overestimate of the L3 cache size.
-constexpr int64_t kL3CacheSizeBytes = 100000000;
-
 // ----------------------------------------------------------------------
 // GPU support
 
@@ -143,22 +140,13 @@ struct ObjectInUseEntry {
   bool is_sealed;
 };
 
-/// Configuration options for the plasma client.
-struct PlasmaClientConfig {
-  /// Number of release calls we wait until the object is actually released.
-  /// This allows us to avoid invalidating the cpu cache on workers if objects
-  /// are reused accross tasks.
-  size_t release_delay;
-};
-
 struct ClientMmapTableEntry {
+  /// The associated file descriptor on the client.
+  int fd;
   /// The result of mmap for this file descriptor.
   uint8_t* pointer;
   /// The length of the memory-mapped file.
   size_t length;
-  /// The number of objects in this memory-mapped file that are currently being
-  /// used by the client. When this count reaches zeros, we unmap the file.
-  int count;
 };
 
 class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Impl> {
@@ -169,7 +157,7 @@ class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Imp
   // PlasmaClient method implementations
 
   Status Connect(const std::string& store_socket_name,
-                 const std::string& manager_socket_name, int release_delay,
+                 const std::string& manager_socket_name, int release_delay = 0,
                  int num_retries = -1);
 
   Status Create(const ObjectID& object_id, int64_t data_size, const uint8_t* metadata,
@@ -210,18 +198,22 @@ class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Imp
 
   Status Disconnect();
 
-  Status FlushReleaseHistory();
-
   bool IsInUse(const ObjectID& object_id);
 
  private:
-  /// This is a helper method for unmapping objects for which all references have
-  /// gone out of scope, either by calling Release or Abort.
+  /// Check if store_fd has already been received from the store. If yes,
+  /// return it. Otherwise, receive it from the store (see analogous logic
+  /// in store.cc).
   ///
-  /// @param object_id The object ID whose data we should unmap.
-  Status UnmapObject(const ObjectID& object_id);
+  /// \param store_fd File descriptor to fetch from the store.
+  /// \return Client file descriptor corresponding to store_fd.
+  int GetStoreFd(int store_fd);
 
-  Status PerformRelease(const ObjectID& object_id);
+  /// This is a helper method for marking an object as unused by this client.
+  ///
+  /// \param object_id The object ID we mark unused.
+  /// \return The return status.
+  Status MarkObjectUnused(const ObjectID& object_id);
 
   /// Common helper for Get() variants
   Status GetBuffers(const ObjectID* object_ids, int64_t num_objects, int64_t timeout_ms,
@@ -254,18 +246,6 @@ class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Imp
   /// A hash table of the object IDs that are currently being used by this
   /// client.
   std::unordered_map<ObjectID, std::unique_ptr<ObjectInUseEntry>> objects_in_use_;
-  /// Object IDs of the last few release calls. This is a deque and
-  /// is used to delay releasing objects to see if they can be reused by
-  /// subsequent tasks so we do not unneccessarily invalidate cpu caches.
-  /// TODO(pcm): replace this with a proper lru cache using the size of the L3
-  /// cache.
-  std::deque<ObjectID> release_history_;
-  /// The number of bytes in the combined objects that are held in the release
-  /// history doubly-linked list. If this is too large then the client starts
-  /// releasing objects.
-  int64_t in_use_object_bytes_;
-  /// Configuration options for the plasma client.
-  PlasmaClientConfig config_;
   /// The amount of memory available to the Plasma store. The client needs this
   /// information to make sure that it does not delay in releasing so much
   /// memory that the store is unable to evict enough objects to free up space.
@@ -295,7 +275,6 @@ PlasmaClient::Impl::~Impl() {}
 uint8_t* PlasmaClient::Impl::LookupOrMmap(int fd, int store_fd_val, int64_t map_size) {
   auto entry = mmap_table_.find(store_fd_val);
   if (entry != mmap_table_.end()) {
-    close(fd);
     return entry->second.pointer;
   } else {
     // We subtract kMmapRegionsGap from the length that was added
@@ -309,9 +288,9 @@ uint8_t* PlasmaClient::Impl::LookupOrMmap(int fd, int store_fd_val, int64_t map_
     close(fd);  // Closing this fd has an effect on performance.
 
     ClientMmapTableEntry& entry = mmap_table_[store_fd_val];
+    entry.fd = fd;
     entry.pointer = result;
     entry.length = map_size;
-    entry.count = 0;
     return result;
   }
 }
@@ -329,6 +308,17 @@ bool PlasmaClient::Impl::IsInUse(const ObjectID& object_id) {
   return (elem != objects_in_use_.end());
 }
 
+int PlasmaClient::Impl::GetStoreFd(int store_fd) {
+  auto entry = mmap_table_.find(store_fd);
+  if (entry == mmap_table_.end()) {
+    int fd = recv_fd(store_conn_);
+    ARROW_CHECK(fd >= 0) << "recv not successful";
+    return fd;
+  } else {
+    return entry->second.fd;
+  }
+}
+
 void PlasmaClient::Impl::IncrementObjectCount(const ObjectID& object_id,
                                               PlasmaObject* object, bool is_sealed) {
   // Increment the count of the object to track the fact that it is being used.
@@ -344,18 +334,6 @@ void PlasmaClient::Impl::IncrementObjectCount(const ObjectID& object_id,
     objects_in_use_[object_id]->count = 0;
     objects_in_use_[object_id]->is_sealed = is_sealed;
     object_entry = objects_in_use_[object_id].get();
-    if (object->device_num == 0) {
-      // Increment the count of the number of objects in the memory-mapped file
-      // that are being used. The corresponding decrement should happen in
-      // PlasmaClient::Release.
-      auto entry = mmap_table_.find(object->store_fd);
-      ARROW_CHECK(entry != mmap_table_.end());
-      ARROW_CHECK(entry->second.count >= 0);
-      // Update the in_use_object_bytes_.
-      in_use_object_bytes_ +=
-          (object_entry->object.data_size + object_entry->object.metadata_size);
-      entry->second.count += 1;
-    }
   } else {
     object_entry = elem->second.get();
     ARROW_CHECK(object_entry->count > 0);
@@ -384,8 +362,7 @@ Status PlasmaClient::Impl::Create(const ObjectID& object_id, int64_t data_size,
   // If the CreateReply included an error, then the store will not send a file
   // descriptor.
   if (device_num == 0) {
-    int fd = recv_fd(store_conn_);
-    ARROW_CHECK(fd >= 0) << "recv not successful";
+    int fd = GetStoreFd(store_fd);
     ARROW_CHECK(object.data_size == data_size);
     ARROW_CHECK(object.metadata_size == metadata_size);
     // The metadata should come right after the data.
@@ -522,8 +499,7 @@ Status PlasmaClient::Impl::GetBuffers(
   // in the subsequent loop based on just the store file descriptor and without
   // having to know the relevant file descriptor received from recv_fd.
   for (size_t i = 0; i < store_fds.size(); i++) {
-    int fd = recv_fd(store_conn_);
-    ARROW_CHECK(fd >= 0);
+    int fd = GetStoreFd(store_fds[i]);
     LookupOrMmap(fd, store_fds[i], mmap_sizes[i]);
   }
 
@@ -602,112 +578,35 @@ Status PlasmaClient::Impl::Get(const ObjectID* object_ids, int64_t num_objects,
   return GetBuffers(object_ids, num_objects, timeout_ms, wrap_buffer, out);
 }
 
-Status PlasmaClient::Impl::UnmapObject(const ObjectID& object_id) {
+Status PlasmaClient::Impl::MarkObjectUnused(const ObjectID& object_id) {
   auto object_entry = objects_in_use_.find(object_id);
   ARROW_CHECK(object_entry != objects_in_use_.end());
   ARROW_CHECK(object_entry->second->count == 0);
 
-  // Decrement the count of the number of objects in this memory-mapped file
-  // that the client is using. The corresponding increment should have
-  // happened in plasma_get.
-  int fd = object_entry->second->object.store_fd;
-  auto entry = mmap_table_.find(fd);
-  ARROW_CHECK(entry != mmap_table_.end());
-  ARROW_CHECK(entry->second.count >= 1);
-  if (entry->second.count == 1) {
-    // If no other objects are being used, then unmap the file.
-    // We subtract kMmapRegionsGap from the length that was added
-    // in fake_mmap in malloc.h, to make the size page-aligned again.
-    int err = munmap(entry->second.pointer, entry->second.length - kMmapRegionsGap);
-    if (err == -1) {
-      return Status::IOError("Error during munmap");
-    }
-    // Remove the corresponding entry from the hash table.
-    mmap_table_.erase(fd);
-  } else {
-    // If there are other objects being used, decrement the reference count.
-    entry->second.count -= 1;
-  }
-  // Update the in_use_object_bytes_.
-  in_use_object_bytes_ -= (object_entry->second->object.data_size +
-                           object_entry->second->object.metadata_size);
-  DCHECK_GE(in_use_object_bytes_, 0);
   // Remove the entry from the hash table of objects currently in use.
   objects_in_use_.erase(object_id);
   return Status::OK();
 }
 
-/// This is a helper method for implementing plasma_release. We maintain a
-/// buffer
-/// of release calls and only perform them once the buffer becomes full (as
-/// judged by the aggregate sizes of the objects). There may be multiple release
-/// calls for the same object ID in the buffer. In this case, the first release
-/// calls will not do anything. The client will only send a message to the store
-/// releasing the object when the client is truly done with the object.
-///
-/// @param object_id The object ID to attempt to release.
-Status PlasmaClient::Impl::PerformRelease(const ObjectID& object_id) {
-  // Decrement the count of the number of instances of this object that are
-  // being used by this client. The corresponding increment should have happened
-  // in PlasmaClient::Get.
-  auto object_entry = objects_in_use_.find(object_id);
-  ARROW_CHECK(object_entry != objects_in_use_.end());
-  object_entry->second->count -= 1;
-  ARROW_CHECK(object_entry->second->count >= 0);
-  // Check if the client is no longer using this object.
-  if (object_entry->second->count == 0) {
-    // Tell the store that the client no longer needs the object.
-    RETURN_NOT_OK(UnmapObject(object_id));
-    RETURN_NOT_OK(SendReleaseRequest(store_conn_, object_id));
-    auto iter = deletion_cache_.find(object_id);
-    if (iter != deletion_cache_.end()) {
-      deletion_cache_.erase(object_id);
-      RETURN_NOT_OK(Delete({object_id}));
-    }
-  }
-  return Status::OK();
-}
-
 Status PlasmaClient::Impl::Release(const ObjectID& object_id) {
   // If the client is already disconnected, ignore release requests.
   if (store_conn_ < 0) {
     return Status::OK();
   }
-  // If an object is in the deletion cache, handle it directly without waiting.
-  auto iter = deletion_cache_.find(object_id);
-  if (iter != deletion_cache_.end()) {
-    RETURN_NOT_OK(PerformRelease(object_id));
-    return Status::OK();
-  }
-  // Add the new object to the release history.
-  release_history_.push_front(object_id);
-  // If there are too many bytes in use by the client or if there are too many
-  // pending release calls, and there are at least some pending release calls in
-  // the release_history list, then release some objects.
-
-  // TODO(wap): Eviction policy only works on host memory, and thus objects on
-  // the GPU cannot be released currently.
-  while ((in_use_object_bytes_ > std::min(kL3CacheSizeBytes, store_capacity_ / 100) ||
-          release_history_.size() > config_.release_delay) &&
-         release_history_.size() > 0) {
-    // Perform a release for the object ID for the first pending release.
-    RETURN_NOT_OK(PerformRelease(release_history_.back()));
-    // Remove the last entry from the release history.
-    release_history_.pop_back();
-  }
-  return Status::OK();
-}
-
-Status PlasmaClient::Impl::FlushReleaseHistory() {
-  // If the client is already disconnected, ignore the flush.
-  if (store_conn_ < 0) {
-    return Status::OK();
-  }
-  while (release_history_.size() > 0) {
-    // Perform a release for the object ID for the first pending release.
-    RETURN_NOT_OK(PerformRelease(release_history_.back()));
-    // Remove the last entry from the release history.
-    release_history_.pop_back();
+  auto object_entry = objects_in_use_.find(object_id);
+  ARROW_CHECK(object_entry != objects_in_use_.end());
+  object_entry->second->count -= 1;
+  ARROW_CHECK(object_entry->second->count >= 0);
+  // Check if the client is no longer using this object.
+  if (object_entry->second->count == 0) {
+    // Tell the store that the client no longer needs the object.
+    RETURN_NOT_OK(MarkObjectUnused(object_id));
+    RETURN_NOT_OK(SendReleaseRequest(store_conn_, object_id));
+    auto iter = deletion_cache_.find(object_id);
+    if (iter != deletion_cache_.end()) {
+      deletion_cache_.erase(object_id);
+      RETURN_NOT_OK(Delete({object_id}));
+    }
   }
   return Status::OK();
 }
@@ -842,8 +741,6 @@ Status PlasmaClient::Impl::Abort(const ObjectID& object_id) {
   ARROW_CHECK(!object_entry->second->is_sealed)
       << "Plasma client called abort on a sealed object";
 
-  // Flush the release history.
-  RETURN_NOT_OK(FlushReleaseHistory());
   // Make sure that the Plasma client only has one reference to the object. If
   // it has more, then the client needs to release the buffer before calling
   // abort.
@@ -855,7 +752,7 @@ Status PlasmaClient::Impl::Abort(const ObjectID& object_id) {
   RETURN_NOT_OK(SendAbortRequest(store_conn_, object_id));
   // Decrease the reference count to zero, then remove the object.
   object_entry->second->count--;
-  RETURN_NOT_OK(UnmapObject(object_id));
+  RETURN_NOT_OK(MarkObjectUnused(object_id));
 
   std::vector<uint8_t> buffer;
   ObjectID id;
@@ -865,7 +762,6 @@ Status PlasmaClient::Impl::Abort(const ObjectID& object_id) {
 }
 
 Status PlasmaClient::Impl::Delete(const std::vector<ObjectID>& object_ids) {
-  RETURN_NOT_OK(FlushReleaseHistory());
   std::vector<ObjectID> not_in_use_ids;
   for (auto& object_id : object_ids) {
     // If the object is in used, skip it.
@@ -965,8 +861,10 @@ Status PlasmaClient::Impl::Connect(const std::string& store_socket_name,
   if (manager_socket_name != "") {
     return Status::NotImplemented("plasma manager is no longer supported");
   }
-  config_.release_delay = release_delay;
-  in_use_object_bytes_ = 0;
+  if (release_delay != 0) {
+    ARROW_LOG(WARNING) << "The release_delay parameter in PlasmaClient::Connect "
+                       << "is deprecated";
+  }
   // Send a ConnectRequest to the store to get its memory capacity.
   RETURN_NOT_OK(SendConnectRequest(store_conn_));
   std::vector<uint8_t> buffer;
@@ -1066,8 +964,6 @@ Status PlasmaClient::DecodeNotification(const uint8_t* buffer, ObjectID* object_
 
 Status PlasmaClient::Disconnect() { return impl_->Disconnect(); }
 
-Status PlasmaClient::FlushReleaseHistory() { return impl_->FlushReleaseHistory(); }
-
 bool PlasmaClient::IsInUse(const ObjectID& object_id) {
   return impl_->IsInUse(object_id);
 }

cpp/src/plasma/client.h

@@ -34,11 +34,6 @@ using arrow::Status;
 
 namespace plasma {
 
-/// We keep a queue of unreleased objects cached in the client until we start
-/// sending release requests to the store. This is to avoid frequently mapping
-/// and unmapping objects and evicting data from processor caches.
-constexpr int64_t kPlasmaDefaultReleaseDelay = 64;
-
 /// Object buffer data structure.
 struct ObjectBuffer {
   /// The data buffer.
@@ -64,13 +59,12 @@ class ARROW_EXPORT PlasmaClient {
   ///        function will not connect to a manager.
   ///        Note that plasma manager is no longer supported, this function
   ///        will return failure if this is not "".
-  /// \param release_delay Number of released objects that are kept around
-  ///        and not evicted to avoid too many munmaps.
+  /// \param release_delay Deprecated (not used).
   /// \param num_retries number of attempts to connect to IPC socket, default 50
   /// \return The return status.
   Status Connect(const std::string& store_socket_name,
-                 const std::string& manager_socket_name,
-                 int release_delay = kPlasmaDefaultReleaseDelay, int num_retries = -1);
+                 const std::string& manager_socket_name, int release_delay = 0,
+                 int num_retries = -1);
 
   /// Create an object in the Plasma Store. Any metadata for this object must be
   /// be passed in when the object is created.
@@ -263,10 +257,6 @@ class ARROW_EXPORT PlasmaClient {
   FRIEND_TEST(TestPlasmaStore, LegacyGetTest);
   FRIEND_TEST(TestPlasmaStore, AbortTest);
 
-  /// This is a helper method that flushes all pending release calls to the
-  /// store.
-  Status FlushReleaseHistory();
-
   bool IsInUse(const ObjectID& object_id);
 
   class ARROW_NO_EXPORT Impl;

cpp/src/plasma/store.cc

@@ -327,7 +327,12 @@ void PlasmaStore::ReturnFromGet(GetRequest* get_req) {
   if (s.ok()) {
     // Send all of the file descriptors for the present objects.
     for (int store_fd : store_fds) {
-      WarnIfSigpipe(send_fd(get_req->client->fd, store_fd), get_req->client->fd);
+      // Only send the file descriptor if it hasn't been sent (see analogous
+      // logic in GetStoreFd in client.cc).
+      if (get_req->client->used_fds.find(store_fd) == get_req->client->used_fds.end()) {
+        WarnIfSigpipe(send_fd(get_req->client->fd, store_fd), get_req->client->fd);
+        get_req->client->used_fds.insert(store_fd);
+      }
     }
   }
 
@@ -783,8 +788,12 @@ Status PlasmaStore::ProcessMessage(Client* client) {
       HANDLE_SIGPIPE(
           SendCreateReply(client->fd, object_id, &object, error_code, mmap_size),
           client->fd);
-      if (error_code == PlasmaError::OK && device_num == 0) {
+      // Only send the file descriptor if it hasn't been sent (see analogous
+      // logic in GetStoreFd in client.cc). Similar in ReturnFromGet.
+      if (error_code == PlasmaError::OK && device_num == 0 &&
+          client->used_fds.find(object.store_fd) == client->used_fds.end()) {
         WarnIfSigpipe(send_fd(client->fd, object.store_fd), client->fd);
+        client->used_fds.insert(object.store_fd);
       }
     } break;
     case fb::MessageType::PlasmaCreateAndSealRequest: {

cpp/src/plasma/store.h

@@ -54,6 +54,9 @@ struct Client {
   /// Object ids that are used by this client.
   std::unordered_set<ObjectID> object_ids;
 
+  /// File descriptors that are used by this client.
+  std::unordered_set<int> used_fds;
+
   /// The file descriptor used to push notifications to client. This is only valid
   /// if client subscribes to plasma store. -1 indicates invalid.
   int notification_fd;

cpp/src/plasma/test/client_tests.cc

@@ -82,7 +82,7 @@ class TestPlasmaStore : public ::testing::Test {
 
   void CreateObject(PlasmaClient& client, const ObjectID& object_id,
                     const std::vector<uint8_t>& metadata,
-                    const std::vector<uint8_t>& data) {
+                    const std::vector<uint8_t>& data, bool release = true) {
     std::shared_ptr<Buffer> data_buffer;
     ARROW_CHECK_OK(client.Create(object_id, data.size(), &metadata[0], metadata.size(),
                                  &data_buffer));
@@ -90,7 +90,9 @@ class TestPlasmaStore : public ::testing::Test {
       data_buffer->mutable_data()[i] = data[i];
     }
     ARROW_CHECK_OK(client.Seal(object_id));
-    ARROW_CHECK_OK(client.Release(object_id));
+    if (release) {
+      ARROW_CHECK_OK(client.Release(object_id));
+    }
   }
 
   const std::string& GetStoreSocketName() const { return store_socket_name_; }
@@ -155,11 +157,12 @@ TEST_F(TestPlasmaStore, SealErrorsTest) {
 
   // Create object.
   std::vector<uint8_t> data(100, 0);
-  CreateObject(client_, object_id, {42}, data);
+  CreateObject(client_, object_id, {42}, data, false);
 
   // Trying to seal it again.
   result = client_.Seal(object_id);
   ASSERT_TRUE(result.IsPlasmaObjectAlreadySealed());
+  ARROW_CHECK_OK(client_.Release(object_id));
 }
 
 TEST_F(TestPlasmaStore, DeleteTest) {
@@ -228,13 +231,7 @@ TEST_F(TestPlasmaStore, DeleteObjectsTest) {
   // client2_ won't send the release request immediately because the trigger
   // condition is not reached. The release is only added to release cache.
   object_buffers.clear();
-  // The reference count went to zero, but the objects are still in the release
-  // cache.
-  ARROW_CHECK_OK(client_.Contains(object_id1, &has_object));
-  ASSERT_TRUE(has_object);
-  ARROW_CHECK_OK(client_.Contains(object_id2, &has_object));
-  ASSERT_TRUE(has_object);
-  // The Delete call will flush release cache and send the Delete request.
+  // Delete the objects.
   result = client2_.Delete(std::vector<ObjectID>{object_id1, object_id2});
   ARROW_CHECK_OK(client_.Contains(object_id1, &has_object));
   ASSERT_FALSE(has_object);
@@ -277,7 +274,6 @@ TEST_F(TestPlasmaStore, GetTest) {
   // First create object.
   std::vector<uint8_t> data = {3, 5, 6, 7, 9};
   CreateObject(client_, object_id, {42}, data);
-  ARROW_CHECK_OK(client_.FlushReleaseHistory());
   EXPECT_FALSE(client_.IsInUse(object_id));
 
   object_buffers.clear();
@@ -291,11 +287,9 @@ TEST_F(TestPlasmaStore, GetTest) {
     auto metadata = object_buffers[0].metadata;
     object_buffers.clear();
     ::arrow::AssertBufferEqual(*metadata, std::string{42});
-    ARROW_CHECK_OK(client_.FlushReleaseHistory());
     EXPECT_TRUE(client_.IsInUse(object_id));
   }
   // Object is automatically released
-  ARROW_CHECK_OK(client_.FlushReleaseHistory());
   EXPECT_FALSE(client_.IsInUse(object_id));
 }
 
@@ -314,17 +308,14 @@ TEST_F(TestPlasmaStore, LegacyGetTest) {
     // First create object.
     std::vector<uint8_t> data = {3, 5, 6, 7, 9};
     CreateObject(client_, object_id, {42}, data);
-    ARROW_CHECK_OK(client_.FlushReleaseHistory());
     EXPECT_FALSE(client_.IsInUse(object_id));
 
     ARROW_CHECK_OK(client_.Get(&object_id, 1, -1, &object_buffer));
     AssertObjectBufferEqual(object_buffer, {42}, {3, 5, 6, 7, 9});
   }
   // Object needs releasing manually
-  ARROW_CHECK_OK(client_.FlushReleaseHistory());
   EXPECT_TRUE(client_.IsInUse(object_id));
   ARROW_CHECK_OK(client_.Release(object_id));
-  ARROW_CHECK_OK(client_.FlushReleaseHistory());
   EXPECT_FALSE(client_.IsInUse(object_id));
 }
 
@@ -377,11 +368,9 @@ TEST_F(TestPlasmaStore, AbortTest) {
   ASSERT_TRUE(status.IsInvalid());
   // Release, then abort.
   ARROW_CHECK_OK(client_.Release(object_id));
-  ARROW_CHECK_OK(client_.FlushReleaseHistory());
   EXPECT_TRUE(client_.IsInUse(object_id));
 
   ARROW_CHECK_OK(client_.Abort(object_id));
-  ARROW_CHECK_OK(client_.FlushReleaseHistory());
   EXPECT_FALSE(client_.IsInUse(object_id));
 
   // Test for object non-existence after the abort.
@@ -394,7 +383,6 @@ TEST_F(TestPlasmaStore, AbortTest) {
   // Test that we can get the object.
   ARROW_CHECK_OK(client_.Get({object_id}, -1, &object_buffers));
   AssertObjectBufferEqual(object_buffers[0], {42, 43}, {1, 2, 3, 4, 5});
-  ARROW_CHECK_OK(client_.Release(object_id));
 }
 
 TEST_F(TestPlasmaStore, MultipleClientTest) {

cpp/tools/parquet/parquet-reader.cc

@@ -23,7 +23,7 @@
 
 int main(int argc, char** argv) {
   if (argc > 5 || argc < 2) {
-    std::cerr << "Usage: parquet_reader [--only-metadata] [--no-memory-map] [--json]"
+    std::cerr << "Usage: parquet-reader [--only-metadata] [--no-memory-map] [--json]"
                  "[--print-key-value-metadata] [--columns=...] <file>"
               << std::endl;
     return -1;

python/pyarrow/_plasma.pyx

@@ -30,10 +30,11 @@ from cython.operator cimport dereference as deref, preincrement as inc
 from cpython.pycapsule cimport *
 
 import collections
-import pyarrow
 import random
 import socket
+import warnings
 
+import pyarrow
 from pyarrow.lib cimport Buffer, NativeFile, check_status, pyarrow_wrap_buffer
 from pyarrow.includes.libarrow cimport (CBuffer, CMutableBuffer,
                                         CFixedSizeBufferWriter, CStatus)
@@ -771,20 +772,22 @@ def connect(store_socket_name, manager_socket_name=None, int release_delay=0,
     store_socket_name : str
         Name of the socket the plasma store is listening at.
     manager_socket_name : str
-        Name of the socket the plasma manager is listening at.
-        This parameter has already been deprecated.
-    release_delay : int, default 0
-        The maximum number of objects that the client will keep and
-        delay releasing (for caching reasons).
+        This parameter is deprecated and has no effect.
+    release_delay : int
+        This parameter is deprecated and has no effect.
     num_retries : int, default -1
         Number of times to try to connect to plasma store. Default value of -1
         uses the default (50)
     """
     if manager_socket_name is not None:
-        print("warning: manager_socket_name parameter for connect() is already deprecated.")
+        warnings.warn("manager_socket_name in PlasmaClient.connect is deprecated",
+                      FutureWarning)
     cdef PlasmaClient result = PlasmaClient()
     result.store_socket_name = store_socket_name.encode()
     result.manager_socket_name = manager_socket_name.encode()
+    if release_delay != 0:
+        warnings.warn("release_delay in PlasmaClient.connect is deprecated",
+                      FutureWarning)
     with nogil:
         check_status(result.client.get()
                      .Connect(result.store_socket_name,

python/pyarrow/table.pxi

@@ -636,12 +636,12 @@ cdef class Column:
 
 cdef _schema_from_arrays(arrays, names, metadata, shared_ptr[CSchema]* schema):
     cdef:
-        Column col
-        c_string c_name
-        vector[shared_ptr[CField]] fields
-        shared_ptr[CDataType] type_
         Py_ssize_t K = len(arrays)
+        c_string c_name
+        CColumn* c_column
+        shared_ptr[CDataType] c_type
         shared_ptr[CKeyValueMetadata] c_meta
+        vector[shared_ptr[CField]] c_fields
 
     if metadata is not None:
         if not isinstance(metadata, dict):
@@ -649,17 +649,15 @@ cdef _schema_from_arrays(arrays, names, metadata, shared_ptr[CSchema]* schema):
         c_meta = pyarrow_unwrap_metadata(metadata)
 
     if K == 0:
-        schema.reset(new CSchema(fields, c_meta))
+        schema.reset(new CSchema(c_fields, c_meta))
         return
 
-    fields.resize(K)
+    c_fields.resize(K)
 
     if isinstance(arrays[0], Column):
         for i in range(K):
-            col = arrays[i]
-            type_ = col.sp_column.get().type()
-            c_name = tobytes(col.name)
-            fields[i].reset(new CField(c_name, type_, True))
+            c_column = (<Column>arrays[i]).column
+            c_fields[i] = c_column.field()
     else:
         if names is None:
             raise ValueError('Must pass names when constructing '
@@ -670,7 +668,7 @@ cdef _schema_from_arrays(arrays, names, metadata, shared_ptr[CSchema]* schema):
         for i in range(K):
             val = arrays[i]
             if isinstance(val, (Array, ChunkedArray)):
-                type_ = (<DataType> val.type).sp_type
+                c_type = (<DataType> val.type).sp_type
             else:
                 raise TypeError(type(val))
 
@@ -678,9 +676,9 @@ cdef _schema_from_arrays(arrays, names, metadata, shared_ptr[CSchema]* schema):
                 c_name = tobytes(u'None')
             else:
                 c_name = tobytes(names[i])
-            fields[i].reset(new CField(c_name, type_, True))
+            c_fields[i].reset(new CField(c_name, c_type, True))
 
-    schema.reset(new CSchema(fields, c_meta))
+    schema.reset(new CSchema(c_fields, c_meta))
 
 
 cdef class RecordBatch:

python/pyarrow/tensorflow/plasma_op.cc

@@ -77,7 +77,7 @@ class TensorToPlasmaOp : public tf::AsyncOpKernel {
     if (!connected_) {
       VLOG(1) << "Connecting to Plasma...";
       ARROW_CHECK_OK(client_.Connect(plasma_store_socket_name_,
-                                     plasma_manager_socket_name_, 0));
+                                     plasma_manager_socket_name_));
       VLOG(1) << "Connected!";
       connected_ = true;
     }
@@ -249,7 +249,7 @@ class PlasmaToTensorOp : public tf::AsyncOpKernel {
     if (!connected_) {
       VLOG(1) << "Connecting to Plasma...";
       ARROW_CHECK_OK(client_.Connect(plasma_store_socket_name_,
-                                     plasma_manager_socket_name_, 0));
+                                     plasma_manager_socket_name_));
       VLOG(1) << "Connected!";
       connected_ = true;
     }

python/pyarrow/tests/test_plasma.py

@@ -121,8 +121,8 @@ class TestPlasmaClient(object):
             use_one_memory_mapped_file=use_one_memory_mapped_file)
         self.plasma_store_name, self.p = self.plasma_store_ctx.__enter__()
         # Connect to Plasma.
-        self.plasma_client = plasma.connect(self.plasma_store_name, "", 64)
-        self.plasma_client2 = plasma.connect(self.plasma_store_name, "", 0)
+        self.plasma_client = plasma.connect(self.plasma_store_name, "")
+        self.plasma_client2 = plasma.connect(self.plasma_store_name, "")
 
     def teardown_method(self, test_method):
         try:
@@ -948,7 +948,7 @@ def test_use_huge_pages():
             plasma_store_memory=2*10**9,
             plasma_directory="/mnt/hugepages",
             use_hugepages=True) as (plasma_store_name, p):
-        plasma_client = plasma.connect(plasma_store_name, "", 64)
+        plasma_client = plasma.connect(plasma_store_name, "")
         create_object(plasma_client, 10**8)
 
 
@@ -962,7 +962,7 @@ def test_plasma_client_sharing():
     with plasma.start_plasma_store(
             plasma_store_memory=DEFAULT_PLASMA_STORE_MEMORY) \
             as (plasma_store_name, p):
-        plasma_client = plasma.connect(plasma_store_name, "", 64)
+        plasma_client = plasma.connect(plasma_store_name, "")
         object_id = plasma_client.put(np.zeros(3))
         buf = plasma_client.get(object_id)
         del plasma_client

python/pyarrow/tests/test_plasma_tf_op.py

@@ -94,7 +94,7 @@ def test_plasma_tf_op(use_gpu=False):
         pytest.skip("TensorFlow Op not found")
 
     with plasma.start_plasma_store(10**8) as (plasma_store_name, p):
-        client = plasma.connect(plasma_store_name, "", 0)
+        client = plasma.connect(plasma_store_name, "")
         for dtype in [np.float32, np.float64,
                       np.int8, np.int16, np.int32, np.int64]:
             run_tensorflow_test_with_dtype(tf, plasma, plasma_store_name,

python/pyarrow/tests/test_table.py

@@ -579,6 +579,21 @@ def test_table_basics():
     assert table.columns == columns
 
 
+def test_table_from_arrays_preserves_column_metadata():
+    # Added to test https://issues.apache.org/jira/browse/ARROW-3866
+    arr0 = pa.array([1, 2])
+    arr1 = pa.array([3, 4])
+    field0 = pa.field('field1', pa.int64(), metadata=dict(a="A", b="B"))
+    field1 = pa.field('field2', pa.int64(), nullable=False)
+    columns = [
+        pa.column(field0, arr0),
+        pa.column(field1, arr1)
+    ]
+    table = pa.Table.from_arrays(columns)
+    assert b"a" in table.column(0).field.metadata
+    assert table.column(1).field.nullable is False
+
+
 def test_table_from_arrays_invalid_names():
     data = [
         pa.array(range(5)),