Merge remote-tracking branch 'remotes/origin/master' into rebase-parquet-datetime

# Conflicts:
#	sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
#	sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala

Author: MaxGekk

R/pkg/NAMESPACE

@@ -345,6 +345,7 @@ exportMethods("%<=>%",
               "over",
               "overlay",
               "percent_rank",
+              "percentile_approx",
               "pmod",
               "posexplode",
               "posexplode_outer",

R/pkg/R/functions.R

@@ -1410,6 +1410,52 @@ setMethod("quarter",
             column(jc)
           })
 
+#' @details
+#' \code{percentile_approx} Returns the approximate percentile value of
+#' numeric column at the given percentage.
+#'
+#' @param percentage Numeric percentage at which percentile should be computed
+#'                   All values should be between 0 and 1.
+#'                   If length equals to 1 resulting column is of type double,
+#'                   otherwise, array type of double.
+#' @param accuracy A positive numeric literal (default: 10000) which
+#'                 controls approximation accuracy at the cost of memory.
+#'                 Higher value of accuracy yields better accuracy, 1.0/accuracy
+#'                 is the relative error of the approximation.
+#'
+#' @rdname column_aggregate_functions
+#' @aliases percentile_approx percentile_approx,Column-method
+#' @note percentile_approx since 3.1.0
+setMethod("percentile_approx",
+          signature(x = "characterOrColumn", percentage = "numericOrColumn"),
+          function(x, percentage, accuracy = 10000) {
+            col <- if (class(x) == "Column") {
+              x@jc
+            } else {
+              column(x)@jc
+            }
+
+            percentage <- if (class(percentage) == "Column") {
+              percentage@jc
+            } else if (length(percentage) > 1) {
+              do.call(create_array, lapply(percentage, lit))@jc
+            } else {
+              lit(percentage)@jc
+            }
+
+            accuracy <- if (class(accuracy) == "Column") {
+              accuracy@jc
+            } else {
+              lit(as.integer(accuracy))@jc
+            }
+
+            jc <- callJStatic(
+              "org.apache.spark.sql.functions", "percentile_approx",
+              col, percentage, accuracy
+            )
+            column(jc)
+          })
+
 #' @details
 #' \code{reverse}: Returns a reversed string or an array with reverse order of elements.
 #'

R/pkg/R/generics.R

@@ -1189,6 +1189,11 @@ setGeneric("overlay", function(x, replace, pos, ...) { standardGeneric("overlay"
 #' @name NULL
 setGeneric("percent_rank", function(x = "missing") { standardGeneric("percent_rank") })
 
+#' @rdname column_aggregate_functions
+#' @name NULL
+setGeneric("percentile_approx",
+           function(x, percentage, ...) { standardGeneric("percentile_approx") })
+
 #' @rdname column_math_functions
 #' @name NULL
 setGeneric("pmod", function(y, x) { standardGeneric("pmod") })

R/pkg/tests/fulltests/test_sparkSQL.R

@@ -1772,6 +1772,28 @@ test_that("column functions", {
     collect(select(df, alias(not(df$is_true), "is_false"))),
     data.frame(is_false = c(FALSE, TRUE, NA))
   )
+
+  # Test percentile_approx
+  actual <- lapply(
+    list(
+      percentile_approx(column("foo"), 0.5),
+      percentile_approx(column("bar"), lit(0.25), lit(42L)),
+      percentile_approx(column("bar"), c(0.25, 0.5, 0.75)),
+      percentile_approx(column("foo"), c(0.05, 0.95), 100L),
+      percentile_approx("foo", c(0.5)),
+      percentile_approx("bar", c(0.1, 0.9), 10L)),
+    function(x) SparkR:::callJMethod(x@jc, "toString"))
+
+  expected <- list(
+     "percentile_approx(foo, 0.5, 10000)",
+     "percentile_approx(bar, 0.25, 42)",
+     "percentile_approx(bar, array(0.25, 0.5, 0.75), 10000)",
+     "percentile_approx(foo, array(0.05, 0.95), 100)",
+     "percentile_approx(foo, 0.5, 10000)",
+     "percentile_approx(bar, array(0.1, 0.9), 10)"
+  )
+
+  expect_equal(actual, expected)
 })
 
 test_that("column binary mathfunctions", {

core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java

@@ -96,8 +96,7 @@ public final class BytesToBytesMap extends MemoryConsumer {
    * since that's the largest power-of-2 that's less than Integer.MAX_VALUE. We need two long array
    * entries per key, giving us a maximum capacity of (1 << 29).
    */
-  @VisibleForTesting
-  static final int MAX_CAPACITY = (1 << 29);
+  public static final int MAX_CAPACITY = (1 << 29);
 
   // This choice of page table size and page size means that we can address up to 500 gigabytes
   // of memory.