Merge branch 'master' into ARROW-3882

cpp/build-support/lintutils.py

@@ -66,13 +66,15 @@ def run_parallel(cmds, **kwargs):
 _source_extensions = '''
 .h
 .cc
+.cpp
 '''.split()
 
 
 def get_sources(source_dir, exclude_globs=[]):
  sources = []
  for directory, subdirs, basenames in os.walk(source_dir):
- for path in [os.path.join(directory, basename) for basename in basenames]:
+ for path in [os.path.join(directory, basename)
+ for basename in basenames]:
  # filter out non-source files
  if os.path.splitext(path)[1] not in _source_extensions:
  continue

cpp/build-support/run_cpplint.py

@@ -110,8 +110,6 @@ def _check_some_files(completed_processes, filenames):
  # distill a list of problematic files
  for problem_files, stdout in pool.imap(checker, chunks):
  if problem_files:
- msg = "{} had cpplint issues"
- print("\n".join(map(msg.format, problem_files)))
  if isinstance(stdout, bytes):
  stdout = stdout.decode('utf8')
  print(stdout, file=sys.stderr)

cpp/src/arrow/array-test.cc

@@ -73,6 +73,10 @@ TEST_F(TestArray, TestNullCount) {
 
  std::unique_ptr<Int32Array> arr_no_nulls(new Int32Array(100, data));
  ASSERT_EQ(0, arr_no_nulls->null_count());
+
+ std::unique_ptr<Int32Array> arr_default_null_count(
+ new Int32Array(100, data, null_bitmap));
+ ASSERT_EQ(kUnknownNullCount, arr_default_null_count->data()->null_count);
 }
 
 TEST_F(TestArray, TestLength) {

cpp/src/arrow/array.h

@@ -378,7 +378,7 @@ class ARROW_EXPORT PrimitiveArray : public FlatArray {
  PrimitiveArray(const std::shared_ptr<DataType>& type, int64_t length,
  const std::shared_ptr<Buffer>& data,
  const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
- int64_t null_count = 0, int64_t offset = 0);
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  /// Does not account for any slice offset
  std::shared_ptr<Buffer> values() const { return data_->buffers[1]; }
@@ -414,8 +414,8 @@ class NumericArray : public PrimitiveArray {
  NumericArray(
  typename std::enable_if<TypeTraits<T1>::is_parameter_free, int64_t>::type length,
  const std::shared_ptr<Buffer>& data,
- const std::shared_ptr<Buffer>& null_bitmap = NULLPTR, int64_t null_count = 0,
- int64_t offset = 0)
+ const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0)
  : PrimitiveArray(TypeTraits<T1>::type_singleton(), length, data, null_bitmap,
  null_count, offset) {}
 
@@ -441,7 +441,7 @@ class ARROW_EXPORT BooleanArray : public PrimitiveArray {
 
  BooleanArray(int64_t length, const std::shared_ptr<Buffer>& data,
  const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
- int64_t null_count = 0, int64_t offset = 0);
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  bool Value(int64_t i) const {
  return BitUtil::GetBit(reinterpret_cast<const uint8_t*>(raw_values_),
@@ -467,8 +467,8 @@ class ARROW_EXPORT ListArray : public Array {
  ListArray(const std::shared_ptr<DataType>& type, int64_t length,
  const std::shared_ptr<Buffer>& value_offsets,
  const std::shared_ptr<Array>& values,
- const std::shared_ptr<Buffer>& null_bitmap = NULLPTR, int64_t null_count = 0,
- int64_t offset = 0);
+ const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  /// \brief Construct ListArray from array of offsets and child value array
  ///
@@ -527,7 +527,7 @@ class ARROW_EXPORT BinaryArray : public FlatArray {
  BinaryArray(int64_t length, const std::shared_ptr<Buffer>& value_offsets,
  const std::shared_ptr<Buffer>& data,
  const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
- int64_t null_count = 0, int64_t offset = 0);
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  /// Return the pointer to the given elements bytes
  // XXX should GetValue(int64_t i) return a string_view?
@@ -585,7 +585,7 @@ class ARROW_EXPORT BinaryArray : public FlatArray {
  const std::shared_ptr<Buffer>& value_offsets,
  const std::shared_ptr<Buffer>& data,
  const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
- int64_t null_count = 0, int64_t offset = 0);
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  const int32_t* raw_value_offsets_;
  const uint8_t* raw_data_;
@@ -601,7 +601,7 @@ class ARROW_EXPORT StringArray : public BinaryArray {
  StringArray(int64_t length, const std::shared_ptr<Buffer>& value_offsets,
  const std::shared_ptr<Buffer>& data,
  const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
- int64_t null_count = 0, int64_t offset = 0);
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 };
 
 // ----------------------------------------------------------------------
@@ -617,7 +617,7 @@ class ARROW_EXPORT FixedSizeBinaryArray : public PrimitiveArray {
  FixedSizeBinaryArray(const std::shared_ptr<DataType>& type, int64_t length,
  const std::shared_ptr<Buffer>& data,
  const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
- int64_t null_count = 0, int64_t offset = 0);
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  const uint8_t* GetValue(int64_t i) const;
  const uint8_t* Value(int64_t i) const { return GetValue(i); }
@@ -673,8 +673,8 @@ class ARROW_EXPORT StructArray : public Array {
 
  StructArray(const std::shared_ptr<DataType>& type, int64_t length,
  const std::vector<std::shared_ptr<Array>>& children,
- std::shared_ptr<Buffer> null_bitmap = NULLPTR, int64_t null_count = 0,
- int64_t offset = 0);
+ std::shared_ptr<Buffer> null_bitmap = NULLPTR,
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  const StructType* struct_type() const;
 
@@ -712,8 +712,8 @@ class ARROW_EXPORT UnionArray : public Array {
  const std::vector<std::shared_ptr<Array>>& children,
  const std::shared_ptr<Buffer>& type_ids,
  const std::shared_ptr<Buffer>& value_offsets = NULLPTR,
- const std::shared_ptr<Buffer>& null_bitmap = NULLPTR, int64_t null_count = 0,
- int64_t offset = 0);
+ const std::shared_ptr<Buffer>& null_bitmap = NULLPTR,
+ int64_t null_count = kUnknownNullCount, int64_t offset = 0);
 
  /// \brief Construct Dense UnionArray from types_ids, value_offsets and children
  ///

cpp/src/arrow/type_traits.h

@@ -43,7 +43,10 @@ struct TypeTraits<NullType> {
  using ArrayType = NullArray;
  using BuilderType = NullBuilder;
  using ScalarType = NullScalar;
- constexpr static bool is_parameter_free = false;
+
+ static constexpr int64_t bytes_required(int64_t) { return 0; }
+ constexpr static bool is_parameter_free = true;
+ static inline std::shared_ptr<DataType> type_singleton() { return null(); }
 };
 
 template <>

cpp/src/plasma/client.cc

@@ -200,6 +200,8 @@ class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Imp
 
  bool IsInUse(const ObjectID& object_id);
 
+ int64_t store_capacity() { return store_capacity_; }
+
  private:
  /// Check if store_fd has already been received from the store. If yes,
  /// return it. Otherwise, receive it from the store (see analogous logic
@@ -968,4 +970,6 @@ bool PlasmaClient::IsInUse(const ObjectID& object_id) {
  return impl_->IsInUse(object_id);
 }
 
+int64_t PlasmaClient::store_capacity() { return impl_->store_capacity(); }
+
 } // namespace plasma

cpp/src/plasma/client.h

@@ -250,6 +250,11 @@ class ARROW_EXPORT PlasmaClient {
  /// \return The return status.
  Status Disconnect();
 
+ /// Get the memory capacity of the store.
+ ///
+ /// \return Memory capacity of the store in bytes.
+ int64_t store_capacity();
+
  private:
  friend class PlasmaBuffer;
  FRIEND_TEST(TestPlasmaStore, GetTest);

cpp/src/plasma/store.cc

@@ -1131,14 +1131,17 @@ int main(int argc, char* argv[]) {
  // shm_vfs_stats.f_bavail is the number of available blocks.
  int64_t shm_mem_avail = shm_vfs_stats.f_bsize * shm_vfs_stats.f_bavail;
  close(shm_fd);
+ // Keep some safety margin for allocator fragmentation.
+ shm_mem_avail = 9 * shm_mem_avail / 10;
  if (system_memory > shm_mem_avail) {
- ARROW_LOG(FATAL)
+ ARROW_LOG(WARNING)
  << "System memory request exceeds memory available in " << plasma_directory
  << ". The request is for " << system_memory
  << " bytes, and the amount available is " << shm_mem_avail
  << " bytes. You may be able to free up space by deleting files in "
  "/dev/shm. If you are inside a Docker container, you may need to "
  "pass an argument with the flag '--shm-size' to 'docker run'.";
+ system_memory = shm_mem_avail;
  }
  } else {
  SetMallocGranularity(1024 * 1024 * 1024); // 1 GB

csharp/Apache.Arrow.sln

@@ -7,6 +7,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow", "src\Apache.
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Tests", "test\Apache.Arrow.Tests\Apache.Arrow.Tests.csproj", "{9CCEC01B-E67A-4726-BE72-7B514F76163F}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Benchmarks", "test\Apache.Arrow.Benchmarks\Apache.Arrow.Benchmarks.csproj", "{742DF47D-77C5-4B84-9E0C-69645F1161EA}"
+EndProject
 Global
  GlobalSection(SolutionConfigurationPlatforms) = preSolution
  Debug|Any CPU = Debug|Any CPU
@@ -21,6 +23,10 @@ Global
  {9CCEC01B-E67A-4726-BE72-7B514F76163F}.Debug|Any CPU.Build.0 = Debug|Any CPU
  {9CCEC01B-E67A-4726-BE72-7B514F76163F}.Release|Any CPU.ActiveCfg = Release|Any CPU
  {9CCEC01B-E67A-4726-BE72-7B514F76163F}.Release|Any CPU.Build.0 = Release|Any CPU
+ {742DF47D-77C5-4B84-9E0C-69645F1161EA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {742DF47D-77C5-4B84-9E0C-69645F1161EA}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {742DF47D-77C5-4B84-9E0C-69645F1161EA}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {742DF47D-77C5-4B84-9E0C-69645F1161EA}.Release|Any CPU.Build.0 = Release|Any CPU
  EndGlobalSection
  GlobalSection(SolutionProperties) = preSolution
  HideSolutionNode = FALSE

csharp/src/Apache.Arrow/Apache.Arrow.csproj

@@ -3,7 +3,7 @@
  <Import Project="../../build/Common.props" />
 
  <PropertyGroup>
- <TargetFramework>netstandard1.3</TargetFramework>
+ <TargetFrameworks>netstandard1.3;netcoreapp2.1</TargetFrameworks>
  <LangVersion>7.2</LangVersion>
  <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
  <Authors>Apache</Authors>
@@ -15,6 +15,7 @@
  <PackageTags>apache arrow</PackageTags>
  <Company>Apache</Company>
  <Version>0.0.1</Version>
+ <DefineConstants>$(DefineConstants);UNSAFE_BYTEBUFFER;BYTEBUFFER_NO_BOUNDS_CHECK;ENABLE_SPAN_T</DefineConstants>
  </PropertyGroup>
 
  <ItemGroup>
@@ -39,4 +40,7 @@
  </EmbeddedResource>
  </ItemGroup>
 
+ <ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp2.1'">
+ <Compile Remove="Extensions\StreamExtensions.netstandard.cs" />
+ </ItemGroup>
 </Project>

csharp/src/Apache.Arrow/ArrowBuffer.cs

@@ -23,7 +23,7 @@ namespace Apache.Arrow
  {
  public static ArrowBuffer Empty => new ArrowBuffer(Memory<byte>.Empty);
 
- private ArrowBuffer(Memory<byte> data)
+ internal ArrowBuffer(ReadOnlyMemory<byte> data)
  {
  Memory = data;
  }

csharp/src/Apache.Arrow/Extensions/StreamExtensions.netstandard.cs

@@ -0,0 +1,54 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.IO;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Apache.Arrow
+{
+ // Helpers to write Memory<byte> to Stream on netstandard
+ internal static class StreamExtensions
+ {
+ public static Task WriteAsync(this Stream stream, ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
+ {
+ if (MemoryMarshal.TryGetArray(buffer, out ArraySegment<byte> array))
+ {
+ return stream.WriteAsync(array.Array, array.Offset, array.Count, cancellationToken);
+ }
+ else
+ {
+ byte[] sharedBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length);
+ buffer.Span.CopyTo(sharedBuffer);
+ return FinishWriteAsync(stream.WriteAsync(sharedBuffer, 0, buffer.Length, cancellationToken), sharedBuffer);
+ }
+ }
+
+ private static async Task FinishWriteAsync(Task writeTask, byte[] localBuffer)
+ {
+ try
+ {
+ await writeTask.ConfigureAwait(false);
+ }
+ finally
+ {
+ ArrayPool<byte>.Shared.Return(localBuffer);
+ }
+ }
+ }
+}