ARROW-11362:[Rust][DataFusion] Use iterator APIs in to_array_of_size to improve performance

.github/workflows/rust.yml

@@ -226,6 +226,41 @@ jobs:
           cd rust
           cargo clippy --all-targets --workspace -- -D warnings -A clippy::redundant_field_names
 
+  miri-checks:
+    name: Miri Checks
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        arch: [amd64]
+        rust: [nightly-2021-01-19]
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: true
+      - uses: actions/cache@v2
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            target
+          key: ${{ runner.os }}-cargo-miri-${{ hashFiles('**/Cargo.lock') }}
+      - name: Setup Rust toolchain
+        run: |
+          rustup toolchain install ${{ matrix.rust }}
+          rustup default ${{ matrix.rust }}
+          rustup component add rustfmt clippy miri
+      - name: Run Miri Checks
+        env:
+          RUST_BACKTRACE: full
+          RUST_LOG: 'trace'
+        run: |
+          export MIRIFLAGS="-Zmiri-disable-isolation"
+          cd rust
+          cargo miri setup
+          cargo clean
+          # Ignore MIRI errors until we can get a clean run
+          cargo miri test || true
+
   coverage:
     name: Coverage
     runs-on: ubuntu-latest

cpp/build-support/fuzzing/generate_corpuses.sh

@@ -42,6 +42,10 @@ rm -rf ${CORPUS_DIR}
 ${OUT}/arrow-ipc-generate-fuzz-corpus -file ${CORPUS_DIR}
 ${ARROW_CPP}/build-support/fuzzing/pack_corpus.py ${CORPUS_DIR} ${OUT}/arrow-ipc-file-fuzz_seed_corpus.zip
 
+rm -rf ${CORPUS_DIR}
+${OUT}/arrow-ipc-generate-tensor-fuzz-corpus -stream ${CORPUS_DIR}
+${ARROW_CPP}/build-support/fuzzing/pack_corpus.py ${CORPUS_DIR} ${OUT}/arrow-ipc-tensor-stream-fuzz_seed_corpus.zip
+
 rm -rf ${CORPUS_DIR}
 ${OUT}/parquet-arrow-generate-fuzz-corpus ${CORPUS_DIR}
 cp ${ARROW_CPP}/submodules/parquet-testing/data/*.parquet ${CORPUS_DIR}

cpp/cmake_modules/SetupCxxFlags.cmake

@@ -267,6 +267,16 @@ if("${BUILD_WARNING_LEVEL}" STREQUAL "CHECKIN")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated-declarations")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-sign-conversion")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-unused-variable")
+  elseif(CMAKE_CXX_COMPILER_ID STREQUAL "Intel")
+    if(WIN32)
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} /Wall")
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} /Wno-deprecated")
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} /Wno-unused-variable")
+    else()
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wall")
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated")
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-unused-variable")
+    endif()
   else()
     message(FATAL_ERROR "${UNKNOWN_COMPILER_MESSAGE}")
   endif()
@@ -289,6 +299,12 @@ elseif("${BUILD_WARNING_LEVEL}" STREQUAL "EVERYTHING")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wpedantic")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wextra")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-unused-parameter")
+  elseif(CMAKE_CXX_COMPILER_ID STREQUAL "Intel")
+    if(WIN32)
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} /Wall")
+    else()
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wall")
+    endif()
   else()
     message(FATAL_ERROR "${UNKNOWN_COMPILER_MESSAGE}")
   endif()
@@ -304,9 +320,14 @@ else()
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} /W3")
   elseif(CMAKE_CXX_COMPILER_ID STREQUAL "AppleClang"
          OR CMAKE_CXX_COMPILER_ID STREQUAL "Clang"
-         OR CMAKE_CXX_COMPILER_ID STREQUAL "GNU"
-         OR CMAKE_CXX_COMPILER_ID STREQUAL "Intel")
+         OR CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wall")
+  elseif(CMAKE_CXX_COMPILER_ID STREQUAL "Intel")
+    if(WIN32)
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} /Wall")
+    else()
+      set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wall")
+    endif()
   else()
     message(FATAL_ERROR "${UNKNOWN_COMPILER_MESSAGE}")
   endif()

cpp/src/arrow/CMakeLists.txt

@@ -365,6 +365,7 @@ if(ARROW_COMPUTE)
               compute/registry.cc
               compute/kernels/aggregate_basic.cc
               compute/kernels/aggregate_mode.cc
+              compute/kernels/aggregate_quantile.cc
               compute/kernels/aggregate_var_std.cc
               compute/kernels/codegen_internal.cc
               compute/kernels/scalar_arithmetic.cc

cpp/src/arrow/compute/api_aggregate.cc

@@ -63,5 +63,10 @@ Result<Datum> Variance(const Datum& value, const VarianceOptions& options,
   return CallFunction("variance", {value}, &options, ctx);
 }
 
+Result<Datum> Quantile(const Datum& value, const QuantileOptions& options,
+                       ExecContext* ctx) {
+  return CallFunction("quantile", {value}, &options, ctx);
+}
+
 }  // namespace compute
 }  // namespace arrow

cpp/src/arrow/compute/api_aggregate.h

@@ -100,6 +100,33 @@ struct ARROW_EXPORT VarianceOptions : public FunctionOptions {
   int ddof = 0;
 };
 
+/// \brief Control Quantile kernel behavior
+///
+/// By default, returns the median value.
+struct ARROW_EXPORT QuantileOptions : public FunctionOptions {
+  /// Interpolation method to use when quantile lies between two data points
+  enum Interpolation {
+    LINEAR = 0,
+    LOWER,
+    HIGHER,
+    NEAREST,
+    MIDPOINT,
+  };
+
+  explicit QuantileOptions(double q = 0.5, enum Interpolation interpolation = LINEAR)
+      : q{q}, interpolation{interpolation} {}
+
+  explicit QuantileOptions(std::vector<double> q,
+                           enum Interpolation interpolation = LINEAR)
+      : q{std::move(q)}, interpolation{interpolation} {}
+
+  static QuantileOptions Defaults() { return QuantileOptions{}; }
+
+  /// quantile must be between 0 and 1 inclusive
+  std::vector<double> q;
+  enum Interpolation interpolation;
+};
+
 /// @}
 
 /// \brief Count non-null (or null) values in an array.
@@ -229,5 +256,19 @@ Result<Datum> Variance(const Datum& value,
                        const VarianceOptions& options = VarianceOptions::Defaults(),
                        ExecContext* ctx = NULLPTR);
 
+/// \brief Calculate the quantiles of a numeric array
+///
+/// \param[in] value input datum, expecting Array or ChunkedArray
+/// \param[in] options see QuantileOptions for more information
+/// \param[in] ctx the function execution context, optional
+/// \return resulting datum as an array
+///
+/// \since 4.0.0
+/// \note API not yet finalized
+ARROW_EXPORT
+Result<Datum> Quantile(const Datum& value,
+                       const QuantileOptions& options = QuantileOptions::Defaults(),
+                       ExecContext* ctx = NULLPTR);
+
 }  // namespace compute
 }  // namespace arrow

cpp/src/arrow/compute/api_scalar.h

@@ -92,6 +92,13 @@ struct ARROW_EXPORT StrptimeOptions : public FunctionOptions {
   TimeUnit::type unit;
 };
 
+struct ARROW_EXPORT TrimOptions : public FunctionOptions {
+  explicit TrimOptions(std::string characters) : characters(std::move(characters)) {}
+
+  /// The individual characters that can be trimmed from the string.
+  std::string characters;
+};
+
 enum CompareOperator : int8_t {
   EQUAL,
   NOT_EQUAL,

cpp/src/arrow/compute/exec.cc

@@ -688,10 +688,9 @@ Status PackBatchNoChunks(const std::vector<Datum>& args, ExecBatch* out) {
     switch (arg.kind()) {
       case Datum::SCALAR:
       case Datum::ARRAY:
+      case Datum::CHUNKED_ARRAY:
         length = std::max(arg.length(), length);
         break;
-      case Datum::CHUNKED_ARRAY:
-        return Status::Invalid("Kernel does not support chunked array arguments");
       default:
         DCHECK(false);
         break;
@@ -722,19 +721,15 @@ class VectorExecutor : public KernelExecutorImpl<VectorKernel> {
                     const std::vector<Datum>& outputs) override {
     // If execution yielded multiple chunks (because large arrays were split
     // based on the ExecContext parameters, then the result is a ChunkedArray
-    if (kernel_->output_chunked) {
-      if (HaveChunkedArray(inputs) || outputs.size() > 1) {
-        return ToChunkedArray(outputs, output_descr_.type);
-      } else if (outputs.size() == 1) {
-        // Outputs have just one element
-        return outputs[0];
-      } else {
-        // XXX: In the case where no outputs are omitted, is returning a 0-length
-        // array always the correct move?
-        return MakeArrayOfNull(output_descr_.type, /*length=*/0).ValueOrDie();
-      }
-    } else {
+    if (kernel_->output_chunked && (HaveChunkedArray(inputs) || outputs.size() > 1)) {
+      return ToChunkedArray(outputs, output_descr_.type);
+    } else if (outputs.size() == 1) {
+      // Outputs have just one element
       return outputs[0];
+    } else {
+      // XXX: In the case where no outputs are omitted, is returning a 0-length
+      // array always the correct move?
+      return MakeArrayOfNull(output_descr_.type, /*length=*/0).ValueOrDie();
     }
   }