Merge branch 'master' into SPARK-10004

Author: Marcelo Vanzin

R/pkg/NAMESPACE

@@ -87,44 +87,86 @@ exportMethods("abs",
               "alias",
               "approxCountDistinct",
               "asc",
+              "ascii",
               "asin",
               "atan",
               "atan2",
               "avg",
+              "base64",
               "between",
+              "bin",
+              "bitwiseNOT",
               "cast",
               "cbrt",
+              "ceil",
               "ceiling",
+              "concat",
               "contains",
               "cos",
               "cosh",
+              "count",
               "countDistinct",
+              "crc32",
+              "datediff",
+              "dayofmonth",
+              "dayofyear",
               "desc",
               "endsWith",
               "exp",
+              "explode",
               "expm1",
+              "factorial",
+              "first",
               "floor",
               "getField",
               "getItem",
+              "greatest",
+              "hex",
+              "hour",
               "hypot",
+              "initcap",
+              "isNaN",
               "isNotNull",
               "isNull",
               "last",
+              "last_day",
+              "least",
+              "length",
+              "levenshtein",
               "like",
+              "lit",
               "log",
               "log10",
               "log1p",
+              "log2",
               "lower",
+              "ltrim",
               "max",
+              "md5",
               "mean",
               "min",
+              "minute",
+              "month",
+              "months_between",
               "n",
               "n_distinct",
+              "nanvl",
+              "negate",
+              "pmod",
+              "quarter",
+              "reverse",
               "rint",
               "rlike",
+              "round",
+              "rtrim",
+              "second",
+              "sha1",
               "sign",
+              "signum",
               "sin",
               "sinh",
+              "size",
+              "soundex",
               "sqrt",
               "startsWith",
               "substr",
@@ -134,7 +176,13 @@ exportMethods("abs",
               "tanh",
               "toDegrees",
               "toRadians",
-              "upper")
+              "to_date",
+              "trim",
+              "unbase64",
+              "unhex",
+              "upper",
+              "weekofyear",
+              "year")
 
 exportClasses("GroupedData")
 exportMethods("agg")

R/pkg/R/deserialize.R

@@ -176,10 +176,14 @@ readRow <- function(inputCon) {
 
 # Take a single column as Array[Byte] and deserialize it into an atomic vector
 readCol <- function(inputCon, numRows) {
-  # sapply can not work with POSIXlt
-  do.call(c, lapply(1:numRows, function(x) {
-    value <- readObject(inputCon)
-    # Replace NULL with NA so we can coerce to vectors
-    if (is.null(value)) NA else value
-  }))
+  if (numRows > 0) {
+    # sapply can not work with POSIXlt
+    do.call(c, lapply(1:numRows, function(x) {
+      value <- readObject(inputCon)
+      # Replace NULL with NA so we can coerce to vectors
+      if (is.null(value)) NA else value
+    }))
+  } else {
+    vector()
+  }
 }

R/pkg/R/functions.R

@@ -67,6 +67,14 @@ createFunctions <- function() {
 
 createFunctions()
 
+#' @rdname functions
+#' @return Creates a Column class of literal value.
+setMethod("lit", signature("ANY"),
+          function(x) {
+            jc <- callJStatic("org.apache.spark.sql.functions", "lit", ifelse(class(x) == "Column", x@jc, x))
+            column(jc)
+          })
+
 #' Approx Count Distinct
 #'
 #' @rdname functions
@@ -93,6 +101,40 @@ setMethod("countDistinct",
             column(jc)
           })
 
+#' @rdname functions
+#' @return Concatenates multiple input string columns together into a single string column.
+setMethod("concat",
+          signature(x = "Column"),
+          function(x, ...) {
+            jcols <- lapply(list(x, ...), function(x) { x@jc })
+            jc <- callJStatic("org.apache.spark.sql.functions", "concat", listToSeq(jcols))
+            column(jc)
+          })
+
+#' @rdname functions
+#' @return Returns the greatest value of the list of column names, skipping null values.
+#'         This function takes at least 2 parameters. It will return null if all parameters are null.
+setMethod("greatest",
+          signature(x = "Column"),
+          function(x, ...) {
+            stopifnot(length(list(...)) > 0)
+            jcols <- lapply(list(x, ...), function(x) { x@jc })
+            jc <- callJStatic("org.apache.spark.sql.functions", "greatest", listToSeq(jcols))
+            column(jc)
+          })
+
+#' @rdname functions
+#' @return Returns the least value of the list of column names, skipping null values.
+#'         This function takes at least 2 parameters. It will return null iff all parameters are null.
+setMethod("least",
+          signature(x = "Column"),
+          function(x, ...) {
+            stopifnot(length(list(...)) > 0)
+            jcols <- lapply(list(x, ...), function(x) { x@jc })
+            jc <- callJStatic("org.apache.spark.sql.functions", "least", listToSeq(jcols))
+            column(jc)
+          })
+
 #' @rdname functions
 #' @aliases ceil
 setMethod("ceiling",

R/pkg/R/generics.R

@@ -682,6 +682,10 @@ setGeneric("cbrt", function(x) { standardGeneric("cbrt") })
 #' @export
 setGeneric("ceil", function(x) { standardGeneric("ceil") })
 
+#' @rdname functions
+#' @export
+setGeneric("concat", function(x, ...) { standardGeneric("concat") })
+
 #' @rdname functions
 #' @export
 setGeneric("crc32", function(x) { standardGeneric("crc32") })
@@ -702,6 +706,10 @@ setGeneric("dayofyear", function(x) { standardGeneric("dayofyear") })
 #' @export
 setGeneric("explode", function(x) { standardGeneric("explode") })
 
+#' @rdname functions
+#' @export
+setGeneric("greatest", function(x, ...) { standardGeneric("greatest") })
+
 #' @rdname functions
 #' @export
 setGeneric("hex", function(x) { standardGeneric("hex") })
@@ -722,10 +730,18 @@ setGeneric("isNaN", function(x) { standardGeneric("isNaN") })
 #' @export
 setGeneric("last_day", function(x) { standardGeneric("last_day") })
 
+#' @rdname functions
+#' @export
+setGeneric("least", function(x, ...) { standardGeneric("least") })
+
 #' @rdname functions
 #' @export
 setGeneric("levenshtein", function(y, x) { standardGeneric("levenshtein") })
 
+#' @rdname functions
+#' @export
+setGeneric("lit", function(x) { standardGeneric("lit") })
+
 #' @rdname functions
 #' @export
 setGeneric("lower", function(x) { standardGeneric("lower") })

R/pkg/inst/tests/test_sparkSQL.R

@@ -408,6 +408,14 @@ test_that("collect() returns a data.frame", {
   expect_equal(names(rdf)[1], "age")
   expect_equal(nrow(rdf), 3)
   expect_equal(ncol(rdf), 2)
+
+  # collect() returns data correctly from a DataFrame with 0 row
+  df0 <- limit(df, 0)
+  rdf <- collect(df0)
+  expect_true(is.data.frame(rdf))
+  expect_equal(names(rdf)[1], "age")
+  expect_equal(nrow(rdf), 0)
+  expect_equal(ncol(rdf), 2)
 })
 
 test_that("limit() returns DataFrame with the correct number of rows", {
@@ -492,6 +500,18 @@ test_that("head() and first() return the correct data", {
 
   testFirst <- first(df)
   expect_equal(nrow(testFirst), 1)
+
+  # head() and first() return the correct data on
+  # a DataFrame with 0 row
+  df0 <- limit(df, 0)
+
+  testHead <- head(df0)
+  expect_equal(nrow(testHead), 0)
+  expect_equal(ncol(testHead), 2)
+
+  testFirst <- first(df0)
+  expect_equal(nrow(testFirst), 0)
+  expect_equal(ncol(testFirst), 2)
 })
 
 test_that("distinct() and unique on DataFrames", {
@@ -560,6 +580,11 @@ test_that("select with column", {
   df2 <- select(df, df$age)
   expect_equal(columns(df2), c("age"))
   expect_equal(count(df2), 3)
+
+  df3 <- select(df, lit("x"))
+  expect_equal(columns(df3), c("x"))
+  expect_equal(count(df3), 3)
+  expect_equal(collect(select(df3, "x"))[[1, 1]], "x")
 })
 
 test_that("selectExpr() on a DataFrame", {
@@ -692,6 +717,14 @@ test_that("string operators", {
   expect_equal(count(where(df, startsWith(df$name, "A"))), 1)
   expect_equal(first(select(df, substr(df$name, 1, 2)))[[1]], "Mi")
   expect_equal(collect(select(df, cast(df$age, "string")))[[2, 1]], "30")
+  expect_equal(collect(select(df, concat(df$name, lit(":"), df$age)))[[2, 1]], "Andy:30")
+})
+
+test_that("greatest() and least() on a DataFrame", {
+  l <- list(list(a = 1, b = 2), list(a = 3, b = 4))
+  df <- createDataFrame(sqlContext, l)
+  expect_equal(collect(select(df, greatest(df$a, df$b)))[, 1], c(2, 4))
+  expect_equal(collect(select(df, least(df$a, df$b)))[, 1], c(1, 3))
 })
 
 test_that("group by", {

conf/log4j.properties.template

@@ -10,6 +10,8 @@ log4j.logger.org.spark-project.jetty=WARN
 log4j.logger.org.spark-project.jetty.util.component.AbstractLifeCycle=ERROR
 log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=INFO
 log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=INFO
+log4j.logger.org.apache.parquet=ERROR
+log4j.logger.parquet=ERROR
 
 # SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
 log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL

core/pom.xml

@@ -266,7 +266,7 @@
     <dependency>
       <groupId>org.tachyonproject</groupId>
       <artifactId>tachyon-client</artifactId>
-      <version>0.7.0</version>
+      <version>0.7.1</version>
       <exclusions>
         <exclusion>
           <groupId>org.apache.hadoop</groupId>

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

@@ -92,9 +92,9 @@ public final class BytesToBytesMap {
 
   /**
    * The maximum number of keys that BytesToBytesMap supports. The hash table has to be
-   * power-of-2-sized and its backing Java array can contain at most (1 << 30) elements, 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).
+   * power-of-2-sized and its backing Java array can contain at most (1 &lt;&lt; 30) elements,
+   * 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 &lt;&lt; 29).
    */
   @VisibleForTesting
   static final int MAX_CAPACITY = (1 << 29);

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -331,6 +331,8 @@ object SparkEnv extends Logging {
         case "netty" =>
           new NettyBlockTransferService(conf, securityManager, numUsableCores)
         case "nio" =>
+          logWarning("NIO-based block transfer service is deprecated, " +
+            "and will be removed in Spark 1.6.0.")
           new NioBlockTransferService(conf, securityManager)
       }
 

core/src/main/scala/org/apache/spark/SparkException.scala

@@ -30,3 +30,10 @@ class SparkException(message: String, cause: Throwable)
  */
 private[spark] class SparkDriverExecutionException(cause: Throwable)
   extends SparkException("Execution error", cause)
+
+/**
+ * Exception thrown when the main user code is run as a child process (e.g. pyspark) and we want
+ * the parent SparkSubmit process to exit with the same exit code.
+ */
+private[spark] case class SparkUserAppException(exitCode: Int)
+  extends SparkException(s"User application exited with $exitCode")