text:
package org.apache.spark.sql

import java.io.CharArrayWriter
import java.util
import java.util.Properties

import scala.language.implicitConversions
import scala.reflect.ClassTag
import scala.reflect.runtime.universe.TypeTag
import scala.util.control.NonFatal
import com.fasterxml.jackson.core.JsonFactory
import org.apache.commons.lang3.StringUtils
import org.apache.spark.annotation.{DeveloperApi, Experimental}
import org.apache.spark.api.java.JavaRDD
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, ScalaReflection, SqlParser}
import org.apache.spark.sql.execution.{EvaluatePython, ExplainCommand, FileRelation, LogicalRDD, SQLExecution}
import org.apache.spark.sql.execution.datasources.{CreateTableUsingAsSelect, LogicalRelation}
import org.apache.spark.sql.execution.datasources.json.JacksonGenerator
import org.apache.spark.sql.sources.HadoopFsRelation
import org.apache.spark.sql.types._
import org.apache.spark.storage.StorageLevel
import org.apache.spark.util.Utils

import scala.util.Random

private[sql] object DataFrame {
  def apply(sqlContext: SQLContext, logicalPlan: LogicalPlan): DataFrame = {
    new DataFrame(sqlContext, logicalPlan)
  }
}

/**
  * :: Experimental ::
  * A distributed collection of data organized into named columns.
  *
  * A [[DataFrame]] is equivalent to a relational table in Spark SQL. The following example creates
  * a [[DataFrame]] by pointing Spark SQL to a Parquet data set.
  * {{{
  *   val people = sqlContext.read.parquet("...")  // in Scala
  *   DataFrame people = sqlContext.read().parquet("...")  // in Java
  * }}}
  *
  * Once created, it can be manipulated using the various domain-specific-language (DSL) functions
  * defined in: [[DataFrame]] (this class), [[Column]], and [[functions]].
  *
  * To select a column from the data frame, use `apply` method in Scala and `col` in Java.
  * {{{
  *   val ageCol = people("age")  // in Scala
  *   Column ageCol = people.col("age")  // in Java
  * }}}
  *
  * Note that the [[Column]] type can also be manipulated through its various functions.
  * {{{
  *   // The following creates a new column that increases everybody's age by 10.
  *   people("age") + 10  // in Scala
  *   people.col("age").plus(10);  // in Java
  * }}}
  *
  * A more concrete example in Scala:
  * {{{
  *   // To create DataFrame using SQLContext
  *   val people = sqlContext.read.parquet("...")
  *   val department = sqlContext.read.parquet("...")
  *
  *   people.filter("age > 30")
  *     .join(department, people("deptId") === department("id"))
  *     .groupBy(department("name"), "gender")
  *     .agg(avg(people("salary")), max(people("age")))
  * }}}
  *
  * and in Java:
  * {{{
  *   // To create DataFrame using SQLContext
  *   DataFrame people = sqlContext.read().parquet("...");
  *   DataFrame department = sqlContext.read().parquet("...");
  *
  *   people.filter("age".gt(30))
  *     .join(department, people.col("deptId").equalTo(department("id")))
  *     .groupBy(department.col("name"), "gender")
  *     .agg(avg(people.col("salary")), max(people.col("age")));
  * }}}
  *
  * @groupname basic Basic DataFrame functions
  * @groupname dfops Language Integrated Queries
  * @groupname rdd RDD Operations
  * @groupname output Output Operations
  * @groupname action Actions
  * @since 1.3.0
  */
// TODO: Improve documentation.
@Experimental
class DataFrame private[sql](@transient val sqlContext: SQLContext,
                             @DeveloperApi @transient val queryExecution: SQLContext#QueryExecution)
  extends Serializable {

  // Note for Spark contributors: if adding or updating any action in `DataFrame`, please make sure
  // you wrap it with `withNewExecutionId` if this actions doesn't call other action.

  /**
    * A constructor that automatically analyzes the logical plan.
    *
    * This reports error eagerly as the [[DataFrame]] is constructed, unless
    * [[SQLConf.dataFrameEagerAnalysis]] is turned off.
    */
  def this(sqlContext: SQLContext, logicalPlan: LogicalPlan) = {
    this(sqlContext, {
      val qe = sqlContext.executePlan(logicalPlan)
      if (sqlContext.conf.dataFrameEagerAnalysis) {
        qe.assertAnalyzed() // This should force analysis and throw errors if there are any
      }
      qe
    })
  }

  // begin of online agg
  // todo set end_confidence/errorBound use conf/confFile
  var end_confidence: Double = 0.99
  var end_errorBound: Double = 0.01

  def setTermination(end_confidence: Double, end_errorBound: Double): Unit = {
    this.end_confidence = end_confidence
    this.end_errorBound = end_errorBound
  }

  // a parameter to some agg
  lazy val size = count()

  /**
    * A onlineAgg access point that can be canceled during execution.
    *
    * by default, the execution will terminate until the sample all
    *
    * cancel condition1: internal & error bound achieve user point
    * cancel condition2: user interactive signal
    *
    * @param groupByCol1       , groupby column name
    * @param aggregateFuncName ,
    *                          aggregate Function name,: avg,sum,count,max,min
    * @param aggregateField    , aggregate used field
    *                          call example: df.onlineAggregate("avg","age","name")
    *                          will calculate DataFrame df's avg of "age" field after groupby "name"
    */
  // todo cancel condition2, "GroupedData's syntax string as parameter"
  // ("onlineAvg('columnName').show()")
  def onlineAggregate(aggregateFuncName: String, aggregateField: String, groupByCol1: String,
                      groupByCols: String*): Unit = {
    // build hashTable for aggre and it's info
    // hashTable

    // sample fraction, stride of sample increase, confidence internal, errorBound
    // above all all agg function specific, so init in online function internal.
    var fraction: Double = 0.01
    var stride: Double = 0.1
    var confidence: Double = 0.6
    var errorBound: Double = 10
    var STRIDE = 0.05
    // todo change stride dynamically
    // var hashTable: util.HashMap[String, Array[Double]] = _

    // i%2 ==0 indicate confidence as parameter
    var i = 0
    while (fraction < 1 && (confidence <= end_confidence || errorBound >= end_errorBound)) {
      //scalastyle:off
      println("sample percentage: " + fraction)
      // scalastyle:on

      val colNames: Seq[String] = groupByCol1 +: groupByCols
      val groupedData = new GroupedData(this.sample(false, fraction),
        colNames.map(colName => resolve(colName)), GroupedData.GroupByType)
      aggregateFuncName match {
        case "avg" =>
          if (i % 2 == 0) {
            var df = groupedData.onlineAvg(confidence, -1d, aggregateField) // .show(false)
            //scalastyle:off
            //println("before update confidence is " + confidence + " er: " + errorBound)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) pair._1 + Random.nextDouble() * STRIDE else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) pair._2 - Random.nextDouble() * STRIDE else 0.01
            //println("after update confidence is " + confidence + " er: " + errorBound)

            // scalastyle:on

          }
          else {
            var df = groupedData.onlineAvg(-1d, errorBound, aggregateField) // .show(false)
            //scalastyle:off
            // println("before update confidence is " + confidence + " er: " + errorBound)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01
            // println("after update confidence is " + confidence + " er: " + errorBound)

            // scalastyle:on
          }

        case "sum" =>
          if (i % 2 == 0) {
            var df = groupedData.onlineSum(confidence, -1d, size, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01

            //            groupedData.onlineSum(confidence, -1d, aggregateField).show(false)
          }
          else {
            var df = groupedData.onlineSum(-1d, errorBound, size, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01
          }
        case "count" =>
          if (i % 2 == 0) {
            var df = groupedData.onlineCount(confidence, -1d, size, fraction, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01

          }
          else {
            var df = groupedData.onlineCount(-1d, errorBound, size, fraction, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01
          }
        case "min" =>
          if (i % 2 == 0) {
            var df = groupedData.onlineMin(confidence, -1d, size, fraction, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01
          }
          else {
            var df = groupedData.onlineMin(-1d, errorBound, size, fraction, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01
          }
        case "max" =>
          if (i % 2 == 0) {
            var df = groupedData.onlineMax(confidence, -1d, size, fraction, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01
          }
          else {
            var df = groupedData.onlineMax(-1d, errorBound, size, fraction, aggregateField)
            val pair = upStatistic(df, confidence, errorBound)
            confidence = if (pair._1 + Random.nextDouble() * STRIDE < 1) {
              pair._1 + Random.nextDouble() * STRIDE
            } else 0.98
            errorBound = if (pair._2 - Random.nextDouble() * STRIDE > 0) {
              pair._2 - Random.nextDouble() * STRIDE
            } else 0.01
          }
        case _ =>
          // scalastyle:off
          println("Invalid aggregate funs")
        // scalastyle:on

      }

      i += 1
      fraction += stride

    }

  }

  /**
    * update online aggregate info, according to field name
    *
    * @param df         the df contains the online agg info
    * @param aggColumns column name after online agg, such as "o?C($col)"
    *                   for Confidence or ErrorBound
    */
  def updateInfo(df: DataFrame, aggColumns: Seq[String]): Seq[Double] = {
    // method1 use df.min agg function
    val result = df.agg(aggColumns.map(col => (col, "min")).toMap).collect()(0)
    result.toSeq.map(field => field.asInstanceOf[Double])

    // var newErrorBound = df.agg(("filed", "min")

  }

  // todo print df schema head, replace print
  def upStatistic(df: DataFrame, con: Double, err: Double): (Double, Double) = {
    var currentConfidence = con
    var currentErrorBound = err

    //scalastyle:off
    println("\n")
    // scalastyle:on
    // Note spark is a distributed plateform, foreach is strange
    //    df.foreach((row: Row) => {
    //      var r: Array[String] = row.toString().split(",")
    //      var numbers = r(r.length - 1).split("\\s+")
    //      //scalastyle:off
    //      val con1 = numbers(1).split("=")(1).toDouble
    //      var err1 = numbers(2).split("=")(1)
    //      val err2 = err1.substring(0, err1.length - 1).toDouble
    //      //println("con1 is: " + con1 + " err: " + err2)
    //      currentConfidence = if (currentConfidence < con1) currentConfidence else con1
    //      //println("after compare con1 is: " + currentConfidence)
    //
    //      currentErrorBound = if (currentErrorBound > err2) currentErrorBound else err2
    //      //println("after compare err is: " + currentErrorBound)
    //
    //      println(row)
    //      // scalastyle:on
    //    }
    //    )
    //scalastyle:off
    //println("curr conf: " + currentConfidence + " err: " + currentErrorBound)
    // scalastyle:on

    val conRdd = df.map((row: Row) => {
      //      println(row)
      //      var r: Array[String] = row.toString().split(",")
      //      var numbers = r(r.length - 1).split("\\s+")
      //      val con1 = numbers(1).split("=")(1).toDouble
      //      var err1 = numbers(2).split("=")(1)
      //      val err2 = err1.substring(0, err1.length - 1).toDouble
      //  new GenericRowWithSchema(Array(con1, err2),
      // StructType(Array(StructField("confidence", DoubleType), StructField("error", DoubleType))))
      var r: Array[String] = row.toString().split(",")
      var numbers = r(r.length - 1).split("\\s+")
      numbers(1).split("=")(1).toDouble

    })
    val errRdd = df.map((row: Row) => {
      //scalastyle:off
      println(row)
      // scalastyle:on
      var r: Array[String] = row.toString().split(",")
      var numbers = r(r.length - 1).split("\\s+")
      var err1 = numbers(2).split("=")(1)
      err1.substring(0, err1.length - 1).toDouble
    })
    //scalastyle:off
    val conMin = conRdd.min()
    val errMin = errRdd.min()
    //    println( conMin + " :con err: " + errMin )
    // scalastyle:on

    (if (currentConfidence < conMin) currentConfidence else conMin,
      if (currentErrorBound > errMin) currentErrorBound else errMin)
  }

  /**
    * update a certain online aggregate info, such as errorBound and confidence
    *
    * @param df     the df contains the online agg info
    * @param aggKey "aggOperation+field"
    * @param index  the index of agg
    */
  def updateOnlineInfo(df: DataFrame, Confidence: Double, ErrorBound: Double,
                       aggKey: String, index: Int): (Double, Double) = {
    // method2, select min confidence, max error
    var currentConfidence = Confidence
    var currentErrorBound = ErrorBound
    // calculate for one pair
    df.map { row =>
      currentConfidence = if (currentConfidence < row.getDouble(8 + index * 4)) currentConfidence
      else row.getDouble(8 + index * 4)
      currentErrorBound = if (currentErrorBound > row.getDouble(16 + index * 4)) currentErrorBound
      else row.getDouble(16 + index * 4)
    }
    (currentConfidence, currentErrorBound)
  }

  @transient protected[sql] val logicalPlan: LogicalPlan = queryExecution.logical match {
    // For various commands (like DDL) and queries with side effects, we force query optimization to
    // happen right away to let these side effects take place eagerly.
    case _: Command |
         _: InsertIntoTable |
         _: CreateTableUsingAsSelect =>
      LogicalRDD(queryExecution.analyzed.output, queryExecution.toRdd)(sqlContext)
    case _ =>
      queryExecution.analyzed
  }

  /**
    * An implicit conversion function internal to this class for us to avoid doing
    * "new DataFrame(...)" everywhere.
    */
  @inline private implicit def logicalPlanToDataFrame(logicalPlan: LogicalPlan): DataFrame = {
    new DataFrame(sqlContext, logicalPlan)
  }

  protected[sql] def resolve(colName: String): NamedExpression = {
    queryExecution.analyzed.resolveQuoted(colName, sqlContext.analyzer.resolver).getOrElse {
      throw new AnalysisException(
        s"""Cannot resolve column name "$colName" among (${schema.fieldNames.mkString(", ")})""")
    }
  }

  protected[sql] def numericColumns: Seq[Expression] = {
    schema.fields.filter(_.dataType.isInstanceOf[NumericType]).map { n =>
      queryExecution.analyzed.resolveQuoted(n.name, sqlContext.analyzer.resolver).get
    }
  }

  /**
    * Compose the string representing rows for output
    *
    * @param _numRows Number of rows to show
    * @param truncate Whether truncate long strings and align cells right
    */
  private[sql] def showString(_numRows: Int, truncate: Boolean = true): String = {
    val numRows = _numRows.max(0)
    val sb = new StringBuilder
    val takeResult = take(numRows + 1)
    val hasMoreData = takeResult.length > numRows
    val data = takeResult.take(numRows)
    val numCols = schema.fieldNames.length

    // For array values, replace Seq and Array with square brackets
    // For cells that are beyond 20 characters, replace it with the first 17 and "..."
    val rows: Seq[Seq[String]] = schema.fieldNames.toSeq +: data.map { row =>
      row.toSeq.map { cell =>
        val str = cell match {
          case null => "null"
          case array: Array[_] => array.mkString("[", ", ", "]")
          case seq: Seq[_] => seq.mkString("[", ", ", "]")
          case _ => cell.toString
        }
        if (truncate && str.length > 20) str.substring(0, 17) + "..." else str
      }: Seq[String]
    }

    // Initialise the width of each column to a minimum value of '3'
    val colWidths = Array.fill(numCols)(3)

    // Compute the width of each column
    for (row <- rows) {
      for ((cell, i) <- row.zipWithIndex) {
        colWidths(i) = math.max(colWidths(i), cell.length)
      }
    }

    // Create SeparateLine
    val sep: String = colWidths.map("-" * _).addString(sb, "+", "+", "+\n").toString()

    // column names
    rows.head.zipWithIndex.map { case (cell, i) =>
      if (truncate) {
        StringUtils.leftPad(cell, colWidths(i))
      } else {
        StringUtils.rightPad(cell, colWidths(i))
      }
    }.addString(sb, "|", "|", "|\n")

    sb.append(sep)

    // data
    rows.tail.map {
      _.zipWithIndex.map { case (cell, i) =>
        if (truncate) {
          StringUtils.leftPad(cell.toString, colWidths(i))
        } else {
          StringUtils.rightPad(cell.toString, colWidths(i))
        }
      }.addString(sb, "|", "|", "|\n")
    }

    sb.append(sep)

    // For Data that has more than "numRows" records
    if (hasMoreData) {
      val rowsString = if (numRows == 1) "row" else "rows"
      sb.append(s"only showing top $numRows ${rowsString}\n")
    }

    sb.toString()
  }

  override def toString: String = {
    try {
      schema.map(f => s"${f.name}: ${f.dataType.simpleString}").mkString("[", ", ", "]")
    } catch {
      case NonFatal(e) =>
        s"Invalid tree; ${e.getMessage}:\n$queryExecution"
    }
  }

  /**
    * Returns the object itself.
    *
    * @group basic
    * @since 1.3.0
    */
  // This is declared with parentheses to prevent the Scala compiler from treating
  // `rdd.toDF("1")` as invoking this toDF and then apply on the returned DataFrame.
  def toDF(): DataFrame = this

  /**
    * Returns a new [[DataFrame]] with columns renamed. This can be quite convenient in conversion
    * from a RDD of tuples into a [[DataFrame]] with meaningful names. For example:
    * {{{
    *   val rdd: RDD[(Int, String)] = ...
    *   rdd.toDF()  // this implicit conversion creates a DataFrame with column name _1 and _2
    *   rdd.toDF("id", "name")  // this creates a DataFrame with column name "id" and "name"
    * }}}
    *
    * @group basic
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def toDF(colNames: String*): DataFrame = {
    require(schema.size == colNames.size,
      "The number of columns doesn't match.\n" +
        s"Old column names (${schema.size}): " + schema.fields.map(_.name).mkString(", ") + "\n" +
        s"New column names (${colNames.size}): " + colNames.mkString(", "))

    val newCols = logicalPlan.output.zip(colNames).map { case (oldAttribute, newName) =>
      Column(oldAttribute).as(newName)
    }
    select(newCols: _*)
  }

  /**
    * Returns the schema of this [[DataFrame]].
    *
    * @group basic
    * @since 1.3.0
    */
  def schema: StructType = queryExecution.analyzed.schema

  /**
    * Returns all column names and their data types as an array.
    *
    * @group basic
    * @since 1.3.0
    */
  def dtypes: Array[(String, String)] = schema.fields.map { field =>
    (field.name, field.dataType.toString)
  }

  /**
    * Returns all column names as an array.
    *
    * @group basic
    * @since 1.3.0
    */
  def columns: Array[String] = schema.fields.map(_.name)

  /**
    * Prints the schema to the console in a nice tree format.
    *
    * @group basic
    * @since 1.3.0
    */
  // scalastyle:off println
  def printSchema(): Unit = println(schema.treeString)

  // scalastyle:on println

  /**
    * Prints the plans (logical and physical) to the console for debugging purposes.
    *
    * @group basic
    * @since 1.3.0
    */
  def explain(extended: Boolean): Unit = {
    val explain = ExplainCommand(queryExecution.logical, extended = extended)
    explain.queryExecution.executedPlan.executeCollect().foreach {
      // scalastyle:off println
      r => println(r.getString(0))
      // scalastyle:on println
    }
  }

  /**
    * Only prints the physical plan to the console for debugging purposes.
    *
    * @group basic
    * @since 1.3.0
    */
  def explain(): Unit = explain(extended = false)

  /**
    * Returns true if the `collect` and `take` methods can be run locally
    * (without any Spark executors).
    *
    * @group basic
    * @since 1.3.0
    */
  def isLocal: Boolean = logicalPlan.isInstanceOf[LocalRelation]

  /**
    * Displays the [[DataFrame]] in a tabular form. Strings more than 20 characters will be
    * truncated, and all cells will be aligned right. For example:
    * {{{
    *   year  month AVG('Adj Close) MAX('Adj Close)
    *   1980  12    0.503218        0.595103
    *   1981  01    0.523289        0.570307
    *   1982  02    0.436504        0.475256
    *   1983  03    0.410516        0.442194
    *   1984  04    0.450090        0.483521
    * }}}
    *
    * @param numRows Number of rows to show
    * @group action
    * @since 1.3.0
    */
  def show(numRows: Int): Unit = show(numRows, true)

  /**
    * Displays the top 20 rows of [[DataFrame]] in a tabular form. Strings more than 20 characters
    * will be truncated, and all cells will be aligned right.
    *
    * @group action
    * @since 1.3.0
    */
  def show(): Unit = show(20)

  /**
    * Displays the top 20 rows of [[DataFrame]] in a tabular form.
    *
    * @param truncate Whether truncate long strings. If true, strings more than 20 characters will
    *                 be truncated and all cells will be aligned right
    * @group action
    * @since 1.5.0
    */
  def show(truncate: Boolean): Unit = show(20, truncate)

  /**
    * Displays the [[DataFrame]] in a tabular form. For example:
    * {{{
    *   year  month AVG('Adj Close) MAX('Adj Close)
    *   1980  12    0.503218        0.595103
    *   1981  01    0.523289        0.570307
    *   1982  02    0.436504        0.475256
    *   1983  03    0.410516        0.442194
    *   1984  04    0.450090        0.483521
    * }}}
    *
    * @param numRows  Number of rows to show
    * @param truncate Whether truncate long strings. If true, strings more than 20 characters will
    *                 be truncated and all cells will be aligned right
    * @group action
    * @since 1.5.0
    */
  // scalastyle:off println
  def show(numRows: Int, truncate: Boolean): Unit = println(showString(numRows, truncate))

  // scalastyle:on println

  /**
    * Returns a [[DataFrameNaFunctions]] for working with missing data.
    * {{{
    *   // Dropping rows containing any null values.
    *   df.na.drop()
    * }}}
    *
    * @group dfops
    * @since 1.3.1
    */
  def na: DataFrameNaFunctions = new DataFrameNaFunctions(this)

  /**
    * Returns a [[DataFrameStatFunctions]] for working statistic functions support.
    * {{{
    *   // Finding frequent items in column with name 'a'.
    *   df.stat.freqItems(Seq("a"))
    * }}}
    *
    * @group dfops
    * @since 1.4.0
    */
  def stat: DataFrameStatFunctions = new DataFrameStatFunctions(this)

  /**
    * Cartesian join with another [[DataFrame]].
    *
    * Note that cartesian joins are very expensive without an extra filter that can be pushed down.
    *
    * @param right Right side of the join operation.
    * @group dfops
    * @since 1.3.0
    */
  def join(right: DataFrame): DataFrame = {
    Join(logicalPlan, right.logicalPlan, joinType = Inner, None)
  }

  /**
    * Inner equi-join with another [[DataFrame]] using the given column.
    *
    * Different from other join functions, the join column will only appear once in the output,
    * i.e. similar to SQL's `JOIN USING` syntax.
    *
    * {{{
    *   // Joining df1 and df2 using the column "user_id"
    *   df1.join(df2, "user_id")
    * }}}
    *
    * Note that if you perform a self-join using this function without aliasing the input
    * [[DataFrame]]s, you will NOT be able to reference any columns after the join, since
    * there is no way to disambiguate which side of the join you would like to reference.
    *
    * @param right       Right side of the join operation.
    * @param usingColumn Name of the column to join on. This column must exist on both sides.
    * @group dfops
    * @since 1.4.0
    */
  def join(right: DataFrame, usingColumn: String): DataFrame = {
    join(right, Seq(usingColumn))
  }

  /**
    * Inner equi-join with another [[DataFrame]] using the given columns.
    *
    * Different from other join functions, the join columns will only appear once in the output,
    * i.e. similar to SQL's `JOIN USING` syntax.
    *
    * {{{
    *   // Joining df1 and df2 using the columns "user_id" and "user_name"
    *   df1.join(df2, Seq("user_id", "user_name"))
    * }}}
    *
    * Note that if you perform a self-join using this function without aliasing the input
    * [[DataFrame]]s, you will NOT be able to reference any columns after the join, since
    * there is no way to disambiguate which side of the join you would like to reference.
    *
    * @param right        Right side of the join operation.
    * @param usingColumns Names of the columns to join on. This columns must exist on both sides.
    * @group dfops
    * @since 1.4.0
    */
  def join(right: DataFrame, usingColumns: Seq[String]): DataFrame = {
    // Analyze the self join. The assumption is that the analyzer will disambiguate left vs right
    // by creating a new instance for one of the branch.
    val joined = sqlContext.executePlan(
      Join(logicalPlan, right.logicalPlan, joinType = Inner, None)).analyzed.asInstanceOf[Join]

    // Project only one of the join columns.
    val joinedCols = usingColumns.map(col => joined.right.resolve(col))
    val condition = usingColumns.map { col =>
      catalyst.expressions.EqualTo(joined.left.resolve(col), joined.right.resolve(col))
    }.reduceLeftOption[catalyst.expressions.BinaryExpression] { (cond, eqTo) =>
      catalyst.expressions.And(cond, eqTo)
    }

    Project(
      joined.output.filterNot(joinedCols.contains(_)),
      Join(
        joined.left,
        joined.right,
        joinType = Inner,
        condition)
    )
  }

  /**
    * Inner join with another [[DataFrame]], using the given join expression.
    *
    * {{{
    *   // The following two are equivalent:
    *   df1.join(df2, $"df1Key" === $"df2Key")
    *   df1.join(df2).where($"df1Key" === $"df2Key")
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def join(right: DataFrame, joinExprs: Column): DataFrame = join(right, joinExprs, "inner")

  /**
    * Join with another [[DataFrame]], using the given join expression. The following performs
    * a full outer join between `df1` and `df2`.
    *
    * {{{
    *   // Scala:
    *   import org.apache.spark.sql.functions._
    *   df1.join(df2, $"df1Key" === $"df2Key", "outer")
    *
    *   // Java:
    *   import static org.apache.spark.sql.functions.*;
    *   df1.join(df2, col("df1Key").equalTo(col("df2Key")), "outer");
    * }}}
    *
    * @param right     Right side of the join.
    * @param joinExprs Join expression.
    * @param joinType  One of: `inner`, `outer`, `left_outer`, `right_outer`, `leftsemi`.
    * @group dfops
    * @since 1.3.0
    */
  def join(right: DataFrame, joinExprs: Column, joinType: String): DataFrame = {
    // Note that in this function, we introduce a hack in the case of self-join to automatically
    // resolve ambiguous join conditions into ones that might make sense [SPARK-6231].
    // Consider this case: df.join(df, df("key") === df("key"))
    // Since df("key") === df("key") is a trivially true condition, this actually becomes a
    // cartesian join. However, most likely users expect to perform a self join using "key".
    // With that assumption, this hack turns the trivially true condition into equality on join
    // keys that are resolved to both sides.

    // Trigger analysis so in the case of self-join, the analyzer will clone the plan.
    // After the cloning, left and right side will have distinct expression ids.
    val plan = Join(logicalPlan, right.logicalPlan, JoinType(joinType), Some(joinExprs.expr))
      .queryExecution.analyzed.asInstanceOf[Join]

    // If auto self join alias is disabled, return the plan.
    if (!sqlContext.conf.dataFrameSelfJoinAutoResolveAmbiguity) {
      return plan
    }

    // If left/right have no output set intersection, return the plan.
    val lanalyzed = this.logicalPlan.queryExecution.analyzed
    val ranalyzed = right.logicalPlan.queryExecution.analyzed
    if (lanalyzed.outputSet.intersect(ranalyzed.outputSet).isEmpty) {
      return plan
    }

    // Otherwise, find the trivially true predicates and automatically resolves them to both sides.
    // By the time we get here, since we have already run analysis, all attributes should've been
    // resolved and become AttributeReference.
    val cond = plan.condition.map {
      _.transform {
        case catalyst.expressions.EqualTo(a: AttributeReference, b: AttributeReference)
          if a.sameRef(b) =>
          catalyst.expressions.EqualTo(plan.left.resolve(a.name), plan.right.resolve(b.name))
      }
    }
    plan.copy(condition = cond)
  }

  /**
    * Returns a new [[DataFrame]] sorted by the specified column, all in ascending order.
    * {{{
    *   // The following 3 are equivalent
    *   df.sort("sortcol")
    *   df.sort($"sortcol")
    *   df.sort($"sortcol".asc)
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def sort(sortCol: String, sortCols: String*): DataFrame = {
    sort((sortCol +: sortCols).map(apply): _*)
  }

  /**
    * Returns a new [[DataFrame]] sorted by the given expressions. For example:
    * {{{
    *   df.sort($"col1", $"col2".desc)
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def sort(sortExprs: Column*): DataFrame = {
    val sortOrder: Seq[SortOrder] = sortExprs.map { col =>
      col.expr match {
        case expr: SortOrder =>
          expr
        case expr: Expression =>
          SortOrder(expr, Ascending)
      }
    }
    Sort(sortOrder, global = true, logicalPlan)
  }

  /**
    * Returns a new [[DataFrame]] sorted by the given expressions.
    * This is an alias of the `sort` function.
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def orderBy(sortCol: String, sortCols: String*): DataFrame = sort(sortCol, sortCols: _*)

  /**
    * Returns a new [[DataFrame]] sorted by the given expressions.
    * This is an alias of the `sort` function.
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def orderBy(sortExprs: Column*): DataFrame = sort(sortExprs: _*)

  /**
    * Selects column based on the column name and return it as a [[Column]].
    * Note that the column name can also reference to a nested column like `a.b`.
    *
    * @group dfops
    * @since 1.3.0
    */
  def apply(colName: String): Column = col(colName)

  /**
    * Selects column based on the column name and return it as a [[Column]].
    * Note that the column name can also reference to a nested column like `a.b`.
    *
    * @group dfops
    * @since 1.3.0
    */
  def col(colName: String): Column = colName match {
    case "*" =>
      Column(ResolvedStar(schema.fieldNames.map(resolve)))
    case _ =>
      val expr = resolve(colName)
      Column(expr)
  }

  /**
    * Returns a new [[DataFrame]] with an alias set.
    *
    * @group dfops
    * @since 1.3.0
    */
  def as(alias: String): DataFrame = Subquery(alias, logicalPlan)

  /**
    * (Scala-specific) Returns a new [[DataFrame]] with an alias set.
    *
    * @group dfops
    * @since 1.3.0
    */
  def as(alias: Symbol): DataFrame = as(alias.name)

  /**
    * Selects a set of column based expressions.
    * {{{
    *   df.select($"colA", $"colB" + 1)
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def select(cols: Column*): DataFrame = {
    val namedExpressions = cols.map {
      // Wrap UnresolvedAttribute with UnresolvedAlias, as when we resolve UnresolvedAttribute, we
      // will remove intermediate Alias for ExtractValue chain, and we need to alias it again to
      // make it a NamedExpression.
      case Column(u: UnresolvedAttribute) => UnresolvedAlias(u)
      case Column(expr: NamedExpression) => expr
      // Leave an unaliased explode with an empty list of names since the analyzer will generate the
      // correct defaults after the nested expression's type has been resolved.
      case Column(explode: Explode) => MultiAlias(explode, Nil)
      case Column(expr: Expression) => Alias(expr, expr.prettyString)()
    }
    Project(namedExpressions.toSeq, logicalPlan)
  }

  /**
    * Selects a set of columns. This is a variant of `select` that can only select
    * existing columns using column names (i.e. cannot construct expressions).
    *
    * {{{
    *   // The following two are equivalent:
    *   df.select("colA", "colB")
    *   df.select($"colA", $"colB")
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def select(col: String, cols: String*): DataFrame = select((col +: cols).map(Column(_)): _*)

  /**
    * Selects a set of SQL expressions. This is a variant of `select` that accepts
    * SQL expressions.
    *
    * {{{
    *   df.selectExpr("colA", "colB as newName", "abs(colC)")
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def selectExpr(exprs: String*): DataFrame = {
    select(exprs.map { expr =>
      Column(SqlParser.parseExpression(expr))
    }: _*)
  }

  /**
    * Filters rows using the given condition.
    * {{{
    *   // The following are equivalent:
    *   peopleDf.filter($"age" > 15)
    *   peopleDf.where($"age" > 15)
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def filter(condition: Column): DataFrame = Filter(condition.expr, logicalPlan)

  /**
    * Filters rows using the given SQL expression.
    * {{{
    *   peopleDf.filter("age > 15")
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def filter(conditionExpr: String): DataFrame = {
    filter(Column(SqlParser.parseExpression(conditionExpr)))
  }

  /**
    * Filters rows using the given condition. This is an alias for `filter`.
    * {{{
    *   // The following are equivalent:
    *   peopleDf.filter($"age" > 15)
    *   peopleDf.where($"age" > 15)
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def where(condition: Column): DataFrame = filter(condition)

  /**
    * Filters rows using the given SQL expression.
    * {{{
    *   peopleDf.where("age > 15")
    * }}}
    *
    * @group dfops
    * @since 1.5.0
    */
  def where(conditionExpr: String): DataFrame = {
    filter(Column(SqlParser.parseExpression(conditionExpr)))
  }

  /**
    * Groups the [[DataFrame]] using the specified columns, so we can run aggregation on them.
    * See [[GroupedData]] for all the available aggregate functions.
    *
    * {{{
    *   // Compute the average for all numeric columns grouped by department.
    *   df.groupBy($"department").avg()
    *
    *   // Compute the max age and average salary, grouped by department and gender.
    *   df.groupBy($"department", $"gender").agg(Map(
    *     "salary" -> "avg",
    *     "age" -> "max"
    *   ))
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def groupBy(cols: Column*): GroupedData = {
    GroupedData(this, cols.map(_.expr), GroupedData.GroupByType)
  }

  /**
    * Create a multi-dimensional rollup for the current [[DataFrame]] using the specified columns,
    * so we can run aggregation on them.
    * See [[GroupedData]] for all the available aggregate functions.
    *
    * {{{
    *   // Compute the average for all numeric columns rolluped by department and group.
    *   df.rollup($"department", $"group").avg()
    *
    *   // Compute the max age and average salary, rolluped by department and gender.
    *   df.rollup($"department", $"gender").agg(Map(
    *     "salary" -> "avg",
    *     "age" -> "max"
    *   ))
    * }}}
    *
    * @group dfops
    * @since 1.4.0
    */
  @scala.annotation.varargs
  def rollup(cols: Column*): GroupedData = {
    GroupedData(this, cols.map(_.expr), GroupedData.RollupType)
  }

  /**
    * Create a multi-dimensional cube for the current [[DataFrame]] using the specified columns,
    * so we can run aggregation on them.
    * See [[GroupedData]] for all the available aggregate functions.
    *
    * {{{
    *   // Compute the average for all numeric columns cubed by department and group.
    *   df.cube($"department", $"group").avg()
    *
    *   // Compute the max age and average salary, cubed by department and gender.
    *   df.cube($"department", $"gender").agg(Map(
    *     "salary" -> "avg",
    *     "age" -> "max"
    *   ))
    * }}}
    *
    * @group dfops
    * @since 1.4.0
    */
  @scala.annotation.varargs
  def cube(cols: Column*): GroupedData = GroupedData(this, cols.map(_.expr), GroupedData.CubeType)

  /**
    * Groups the [[DataFrame]] using the specified columns, so we can run aggregation on them.
    * See [[GroupedData]] for all the available aggregate functions.
    *
    * This is a variant of groupBy that can only group by existing columns using column names
    * (i.e. cannot construct expressions).
    *
    * {{{
    *   // Compute the average for all numeric columns grouped by department.
    *   df.groupBy("department").avg()
    *
    *   // Compute the max age and average salary, grouped by department and gender.
    *   df.groupBy($"department", $"gender").agg(Map(
    *     "salary" -> "avg",
    *     "age" -> "max"
    *   ))
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def groupBy(col1: String, cols: String*): GroupedData = {
    val colNames: Seq[String] = col1 +: cols
    GroupedData(this, colNames.map(colName => resolve(colName)), GroupedData.GroupByType)
  }

  /**
    * Create a multi-dimensional rollup for the current [[DataFrame]] using the specified columns,
    * so we can run aggregation on them.
    * See [[GroupedData]] for all the available aggregate functions.
    *
    * This is a variant of rollup that can only group by existing columns using column names
    * (i.e. cannot construct expressions).
    *
    * {{{
    *   // Compute the average for all numeric columns rolluped by department and group.
    *   df.rollup("department", "group").avg()
    *
    *   // Compute the max age and average salary, rolluped by department and gender.
    *   df.rollup($"department", $"gender").agg(Map(
    *     "salary" -> "avg",
    *     "age" -> "max"
    *   ))
    * }}}
    *
    * @group dfops
    * @since 1.4.0
    */
  @scala.annotation.varargs
  def rollup(col1: String, cols: String*): GroupedData = {
    val colNames: Seq[String] = col1 +: cols
    GroupedData(this, colNames.map(colName => resolve(colName)), GroupedData.RollupType)
  }

  /**
    * Create a multi-dimensional cube for the current [[DataFrame]] using the specified columns,
    * so we can run aggregation on them.
    * See [[GroupedData]] for all the available aggregate functions.
    *
    * This is a variant of cube that can only group by existing columns using column names
    * (i.e. cannot construct expressions).
    *
    * {{{
    *   // Compute the average for all numeric columns cubed by department and group.
    *   df.cube("department", "group").avg()
    *
    *   // Compute the max age and average salary, cubed by department and gender.
    *   df.cube($"department", $"gender").agg(Map(
    *     "salary" -> "avg",
    *     "age" -> "max"
    *   ))
    * }}}
    *
    * @group dfops
    * @since 1.4.0
    */
  @scala.annotation.varargs
  def cube(col1: String, cols: String*): GroupedData = {
    val colNames: Seq[String] = col1 +: cols
    GroupedData(this, colNames.map(colName => resolve(colName)), GroupedData.CubeType)
  }

  /**
    * (Scala-specific) Aggregates on the entire [[DataFrame]] without groups.
    * {{{
    *   // df.agg(...) is a shorthand for df.groupBy().agg(...)
    *   df.agg("age" -> "max", "salary" -> "avg")
    *   df.groupBy().agg("age" -> "max", "salary" -> "avg")
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def agg(aggExpr: (String, String), aggExprs: (String, String)*): DataFrame = {
    groupBy().agg(aggExpr, aggExprs: _*)
  }

  /**
    * (Scala-specific) Aggregates on the entire [[DataFrame]] without groups.
    * {{{
    *   // df.agg(...) is a shorthand for df.groupBy().agg(...)
    *   df.agg(Map("age" -> "max", "salary" -> "avg"))
    *   df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def agg(exprs: Map[String, String]): DataFrame = groupBy().agg(exprs)

  /**
    * (Java-specific) Aggregates on the entire [[DataFrame]] without groups.
    * {{{
    *   // df.agg(...) is a shorthand for df.groupBy().agg(...)
    *   df.agg(Map("age" -> "max", "salary" -> "avg"))
    *   df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def agg(exprs: java.util.Map[String, String]): DataFrame = groupBy().agg(exprs)

  /**
    * Aggregates on the entire [[DataFrame]] without groups.
    * {{{
    *   // df.agg(...) is a shorthand for df.groupBy().agg(...)
    *   df.agg(max($"age"), avg($"salary"))
    *   df.groupBy().agg(max($"age"), avg($"salary"))
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  @scala.annotation.varargs
  def agg(expr: Column, exprs: Column*): DataFrame = groupBy().agg(expr, exprs: _*)

  /**
    * Returns a new [[DataFrame]] by taking the first `n` rows. The difference between this function
    * and `head` is that `head` returns an array while `limit` returns a new [[DataFrame]].
    *
    * @group dfops
    * @since 1.3.0
    */
  def limit(n: Int): DataFrame = Limit(Literal(n), logicalPlan)

  /**
    * Returns a new [[DataFrame]] containing union of rows in this frame and another frame.
    * This is equivalent to `UNION ALL` in SQL.
    *
    * @group dfops
    * @since 1.3.0
    */
  def unionAll(other: DataFrame): DataFrame = Union(logicalPlan, other.logicalPlan)

  /**
    * Returns a new [[DataFrame]] containing rows only in both this frame and another frame.
    * This is equivalent to `INTERSECT` in SQL.
    *
    * @group dfops
    * @since 1.3.0
    */
  def intersect(other: DataFrame): DataFrame = Intersect(logicalPlan, other.logicalPlan)

  /**
    * Returns a new [[DataFrame]] containing rows in this frame but not in another frame.
    * This is equivalent to `EXCEPT` in SQL.
    *
    * @group dfops
    * @since 1.3.0
    */
  def except(other: DataFrame): DataFrame = Except(logicalPlan, other.logicalPlan)

  /**
    * Returns a new [[DataFrame]] by sampling a fraction of rows.
    *
    * @param withReplacement Sample with replacement or not.
    * @param fraction        Fraction of rows to generate.
    * @param seed            Seed for sampling.
    * @group dfops
    * @since 1.3.0
    */
  def sample(withReplacement: Boolean, fraction: Double, seed: Long): DataFrame = {
    Sample(0.0, fraction, withReplacement, seed, logicalPlan)
  }

  /**
    * Returns a new [[DataFrame]] by sampling a fraction of rows, using a random seed.
    *
    * @param withReplacement Sample with replacement or not.
    * @param fraction        Fraction of rows to generate.
    * @group dfops
    * @since 1.3.0
    */
  def sample(withReplacement: Boolean, fraction: Double): DataFrame = {
    sample(withReplacement, fraction, Utils.random.nextLong)
  }

  /**
    * Randomly splits this [[DataFrame]] with the provided weights.
    *
    * @param weights weights for splits, will be normalized if they don't sum to 1.
    * @param seed    Seed for sampling.
    * @group dfops
    * @since 1.4.0
    */
  def randomSplit(weights: Array[Double], seed: Long): Array[DataFrame] = {
    val sum = weights.sum
    val normalizedCumWeights = weights.map(_ / sum).scanLeft(0.0d)(_ + _)
    normalizedCumWeights.sliding(2).map { x =>
      new DataFrame(sqlContext, Sample(x(0), x(1), false, seed, logicalPlan))
    }.toArray
  }

  /**
    * Randomly splits this [[DataFrame]] with the provided weights.
    *
    * @param weights weights for splits, will be normalized if they don't sum to 1.
    * @group dfops
    * @since 1.4.0
    */
  def randomSplit(weights: Array[Double]): Array[DataFrame] = {
    randomSplit(weights, Utils.random.nextLong)
  }

  /**
    * Randomly splits this [[DataFrame]] with the provided weights. Provided for the Python Api.
    *
    * @param weights weights for splits, will be normalized if they don't sum to 1.
    * @param seed    Seed for sampling.
    * @group dfops
    */
  private[spark] def randomSplit(weights: List[Double], seed: Long): Array[DataFrame] = {
    randomSplit(weights.toArray, seed)
  }

  /**
    * (Scala-specific) Returns a new [[DataFrame]] where each row has been expanded to zero or more
    * rows by the provided function.  This is similar to a `LATERAL VIEW` in HiveQL. The columns of
    * the input row are implicitly joined with each row that is output by the function.
    *
    * The following example uses this function to count the number of books which contain
    * a given word:
    *
    * {{{
    *   case class Book(title: String, words: String)
    *   val df: RDD[Book]
    *
    *   case class Word(word: String)
    *   val allWords = df.explode('words) {
    *     case Row(words: String) => words.split(" ").map(Word(_))
    *   }
    *
    *   val bookCountPerWord = allWords.groupBy("word").agg(countDistinct("title"))
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def explode[A <: Product : TypeTag](input: Column*)(f: Row => TraversableOnce[A]): DataFrame = {
    val schema = ScalaReflection.schemaFor[A].dataType.asInstanceOf[StructType]

    val elementTypes = schema.toAttributes.map { attr => (attr.dataType, attr.nullable) }
    val names = schema.toAttributes.map(_.name)
    val convert = CatalystTypeConverters.createToCatalystConverter(schema)

    val rowFunction =
      f.andThen(_.map(convert(_).asInstanceOf[InternalRow]))
    val generator = UserDefinedGenerator(elementTypes, rowFunction, input.map(_.expr))

    Generate(generator, join = true, outer = false,
      qualifier = None, names.map(UnresolvedAttribute(_)), logicalPlan)
  }

  /**
    * (Scala-specific) Returns a new [[DataFrame]] where a single column has been expanded to zero
    * or more rows by the provided function.  This is similar to a `LATERAL VIEW` in HiveQL. All
    * columns of the input row are implicitly joined with each value that is output by the function.
    *
    * {{{
    *   df.explode("words", "word"){words: String => words.split(" ")}
    * }}}
    *
    * @group dfops
    * @since 1.3.0
    */
  def explode[A, B: TypeTag](inputColumn: String, outputColumn: String)(f: A => TraversableOnce[B])
  : DataFrame = {
    val dataType = ScalaReflection.schemaFor[B].dataType
    val attributes = AttributeReference(outputColumn, dataType)() :: Nil
    // TODO handle the metadata?
    val elementTypes = attributes.map { attr => (attr.dataType, attr.nullable) }
    val names = attributes.map(_.name)

    def rowFunction(row: Row): TraversableOnce[InternalRow] = {
      val convert = CatalystTypeConverters.createToCatalystConverter(dataType)
      f(row(0).asInstanceOf[A]).map(o => InternalRow(convert(o)))
    }

    val generator = UserDefinedGenerator(elementTypes, rowFunction, apply(inputColumn).expr :: Nil)

    Generate(generator, join = true, outer = false,
      qualifier = None, names.map(UnresolvedAttribute(_)), logicalPlan)
  }

  /////////////////////////////////////////////////////////////////////////////

  /**
    * Returns a new [[DataFrame]] by adding a column or replacing the existing column that has
    * the same name.
    *
    * @group dfops
    * @since 1.3.0
    */
  def withColumn(colName: String, col: Column): DataFrame = {
    val resolver = sqlContext.analyzer.resolver
    val replaced = schema.exists(f => resolver(f.name, colName))
    if (replaced) {
      val colNames = schema.map { field =>
        val name = field.name
        if (resolver(name, colName)) col.as(colName) else Column(name)
      }
      select(colNames: _*)
    } else {
      select(Column("*"), col.as(colName))
    }
  }

  /**
    * Returns a new [[DataFrame]] with a column renamed.
    * This is a no-op if schema doesn't contain existingName.
    *
    * @group dfops
    * @since 1.3.0
    */
  def withColumnRenamed(existingName: String, newName: String): DataFrame = {
    val resolver = sqlContext.analyzer.resolver
    val shouldRename = schema.exists(f => resolver(f.name, existingName))
    if (shouldRename) {
      val colNames = schema.map { field =>
        val name = field.name
        if (resolver(name, existingName)) Column(name).as(newName) else Column(name)
      }
      select(colNames: _*)
    } else {
      this
    }
  }

  /**
    * Returns a new [[DataFrame]] with a column dropped.
    * This is a no-op if schema doesn't contain column name.
    *
    * @group dfops
    * @since 1.4.0
    */
  def drop(colName: String): DataFrame = {
    val resolver = sqlContext.analyzer.resolver
    val shouldDrop = schema.exists(f => resolver(f.name, colName))
    if (shouldDrop) {
      val colsAfterDrop = schema.filter { field =>
        val name = field.name
        !resolver(name, colName)
      }.map(f => Column(f.name))
      select(colsAfterDrop: _*)
    } else {
      this
    }
  }

  /**
    * Returns a new [[DataFrame]] with a column dropped.
    * This version of drop accepts a Column rather than a name.
    * This is a no-op if the DataFrame doesn't have a column
    * with an equivalent expression.
    *
    * @group dfops
    * @since 1.4.1
    */
  def drop(col: Column): DataFrame = {
    val attrs = this.logicalPlan.output
    val colsAfterDrop = attrs.filter { attr =>
      attr != col.expr
    }.map(attr => Column(attr))
    select(colsAfterDrop: _*)
  }

  /**
    * Returns a new [[DataFrame]] that contains only the unique rows from this [[DataFrame]].
    * This is an alias for `distinct`.
    *
    * @group dfops
    * @since 1.4.0
    */
  def dropDuplicates(): DataFrame = dropDuplicates(this.columns)

  /**
    * (Scala-specific) Returns a new [[DataFrame]] with duplicate rows removed, considering only
    * the subset of columns.
    *
    * @group dfops
    * @since 1.4.0
    */
  def dropDuplicates(colNames: Seq[String]): DataFrame = {
    val groupCols = colNames.map(resolve)
    val groupColExprIds = groupCols.map(_.exprId)
    val aggCols = logicalPlan.output.map { attr =>
      if (groupColExprIds.contains(attr.exprId)) {
        attr
      } else {
        Alias(First(attr), attr.name)()
      }
    }
    Aggregate(groupCols, aggCols, logicalPlan)
  }

  /**
    * Returns a new [[DataFrame]] with duplicate rows removed, considering only
    * the subset of columns.
    *
    * @group dfops
    * @since 1.4.0
    */
  def dropDuplicates(colNames: Array[String]): DataFrame = dropDuplicates(colNames.toSeq)

  /**
    * Computes statistics for numeric columns, including count, mean, stddev, min, and max.
    * If no columns are given, this function computes statistics for all numerical columns.
    *
    * This function is meant for exploratory data analysis, as we make no guarantee about the
    * backward compatibility of the schema of the resulting [[DataFrame]]. If you want to
    * programmatically compute summary statistics, use the `agg` function instead.
    *
    * {{{
    *   df.describe("age", "height").show()
    *
    *   // output:
    *   // summary age   height
    *   // count   10.0  10.0
    *   // mean    53.3  178.05
    *   // stddev  11.6  15.7
    *   // min     18.0  163.0
    *   // max     92.0  192.0
    * }}}
    *
    * @group action
    * @since 1.3.1
    */
  @scala.annotation.varargs
  def describe(cols: String*): DataFrame = {

    // TODO: Add stddev as an expression, and remove it from here.
    def stddevExpr(expr: Expression): Expression =
    Sqrt(Subtract(Average(Multiply(expr, expr)), Multiply(Average(expr), Average(expr))))

    // The list of summary statistics to compute, in the form of expressions.
    val statistics = List[(String, Expression => Expression)](
      "count" -> Count,
      "mean" -> Average,
      "stddev" -> stddevExpr,
      "min" -> Min,
      "max" -> Max)

    val outputCols = (if (cols.isEmpty) numericColumns.map(_.prettyString) else cols).toList

    val ret: Seq[Row] = if (outputCols.nonEmpty) {
      val aggExprs = statistics.flatMap { case (_, colToAgg) =>
        outputCols.map(c => Column(Cast(colToAgg(Column(c).expr), StringType)).as(c))
      }

      val row = agg(aggExprs.head, aggExprs.tail: _*).head().toSeq

      // Pivot the data so each summary is one row
      row.grouped(outputCols.size).toSeq.zip(statistics).map { case (aggregation, (statistic, _)) =>
        Row(statistic :: aggregation.toList: _*)
      }
    } else {
      // If there are no output columns, just output a single column that contains the stats.
      statistics.map { case (name, _) => Row(name) }
    }

    // All columns are string type
    val schema = StructType(
      StructField("summary", StringType) :: outputCols.map(StructField(_, StringType))).toAttributes
    LocalRelation.fromExternalRows(schema, ret)
  }

  /**
    * Returns the first `n` rows.
    *
    * @group action
    * @since 1.3.0
    */
  def head(n: Int): Array[Row] = limit(n).collect()

  /**
    * Returns the first row.
    *
    * @group action
    * @since 1.3.0
    */
  def head(): Row = head(1).head

  /**
    * Returns the first row. Alias for head().
    *
    * @group action
    * @since 1.3.0
    */
  def first(): Row = head()

  /**
    * Returns a new RDD by applying a function to all rows of this DataFrame.
    *
    * @group rdd
    * @since 1.3.0
    */
  def map[R: ClassTag](f: Row => R): RDD[R] = rdd.map(f)

  /**
    * Returns a new RDD by first applying a function to all rows of this [[DataFrame]],
    * and then flattening the results.
    *
    * @group rdd
    * @since 1.3.0
    */
  def flatMap[R: ClassTag](f: Row => TraversableOnce[R]): RDD[R] = rdd.flatMap(f)

  /**
    * Returns a new RDD by applying a function to each partition of this DataFrame.
    *
    * @group rdd
    * @since 1.3.0
    */
  def mapPartitions[R: ClassTag](f: Iterator[Row] => Iterator[R]): RDD[R] = {
    rdd.mapPartitions(f)
  }

  /**
    * Applies a function `f` to all rows.
    *
    * @group rdd
    * @since 1.3.0
    */
  def foreach(f: Row => Unit): Unit = withNewExecutionId {
    rdd.foreach(f)
  }

  /**
    * Applies a function f to each partition of this [[DataFrame]].
    *
    * @group rdd
    * @since 1.3.0
    */
  def foreachPartition(f: Iterator[Row] => Unit): Unit = withNewExecutionId {
    rdd.foreachPartition(f)
  }

  /**
    * Returns the first `n` rows in the [[DataFrame]].
    *
    * @group action
    * @since 1.3.0
    */
  def take(n: Int): Array[Row] = head(n)

  /**
    * Returns an array that contains all of [[Row]]s in this [[DataFrame]].
    *
    * @group action
    * @since 1.3.0
    */
  def collect(): Array[Row] = withNewExecutionId {
    queryExecution.executedPlan.executeCollect()
  }

  /**
    * Returns a Java list that contains all of [[Row]]s in this [[DataFrame]].
    *
    * @group action
    * @since 1.3.0
    */
  def collectAsList(): java.util.List[Row] = withNewExecutionId {
    java.util.Arrays.asList(rdd.collect(): _*)
  }

  /**
    * Returns the number of rows in the [[DataFrame]].
    *
    * @group action
    * @since 1.3.0
    */
  def count(): Long = groupBy().count().collect().head.getLong(0)

  /**
    * Returns a new [[DataFrame]] that has exactly `numPartitions` partitions.
    *
    * @group rdd
    * @since 1.3.0
    */
  def repartition(numPartitions: Int): DataFrame = {
    Repartition(numPartitions, shuffle = true, logicalPlan)
  }

  /**
    * Returns a new [[DataFrame]] that has exactly `numPartitions` partitions.
    * Similar to coalesce defined on an [[RDD]], this operation results in a narrow dependency, e.g.
    * if you go from 1000 partitions to 100 partitions, there will not be a shuffle, instead each of
    * the 100 new partitions will claim 10 of the current partitions.
    *
    * @group rdd
    * @since 1.4.0
    */
  def coalesce(numPartitions: Int): DataFrame = {
    Repartition(numPartitions, shuffle = false, logicalPlan)
  }

  /**
    * Returns a new [[DataFrame]] that contains only the unique rows from this [[DataFrame]].
    * This is an alias for `dropDuplicates`.
    *
    * @group dfops
    * @since 1.3.0
    */
  def distinct(): DataFrame = dropDuplicates()

  /**
    * @group basic
    * @since 1.3.0
    */
  def persist(): this.type = {
    sqlContext.cacheManager.cacheQuery(this)
    this
  }

  /**
    * @group basic
    * @since 1.3.0
    */
  def cache(): this.type = persist()

  /**
    * @group basic
    * @since 1.3.0
    */
  def persist(newLevel: StorageLevel): this.type = {
    sqlContext.cacheManager.cacheQuery(this, None, newLevel)
    this
  }

  /**
    * @group basic
    * @since 1.3.0
    */
  def unpersist(blocking: Boolean): this.type = {
    sqlContext.cacheManager.tryUncacheQuery(this, blocking)
    this
  }

  /**
    * @group basic
    * @since 1.3.0
    */
  def unpersist(): this.type = unpersist(blocking = false)

  /////////////////////////////////////////////////////////////////////////////
  // I/O
  /////////////////////////////////////////////////////////////////////////////

  /**
    * Represents the content of the [[DataFrame]] as an [[RDD]] of [[Row]]s. Note that the RDD is
    * memoized. Once called, it won't change even if you change any query planning related Spark SQL
    * configurations (e.g. `spark.sql.shuffle.partitions`).
    *
    * @group rdd
    * @since 1.3.0
    */
  lazy val rdd: RDD[Row] = {
    // use a local variable to make sure the map closure doesn't capture the whole DataFrame
    val schema = this.schema
    queryExecution.toRdd.mapPartitions { rows =>
      val converter = CatalystTypeConverters.createToScalaConverter(schema)
      rows.map(converter(_).asInstanceOf[Row])
    }
  }

  /**
    * Returns the content of the [[DataFrame]] as a [[JavaRDD]] of [[Row]]s.
    *
    * @group rdd
    * @since 1.3.0
    */
  def toJavaRDD: JavaRDD[Row] = rdd.toJavaRDD()

  /**
    * Returns the content of the [[DataFrame]] as a [[JavaRDD]] of [[Row]]s.
    *
    * @group rdd
    * @since 1.3.0
    */
  def javaRDD: JavaRDD[Row] = toJavaRDD

  /**
    * Registers this [[DataFrame]] as a temporary table using the given name.  The lifetime of this
    * temporary table is tied to the [[SQLContext]] that was used to create this DataFrame.
    *
    * @group basic
    * @since 1.3.0
    */
  def registerTempTable(tableName: String): Unit = {
    sqlContext.registerDataFrameAsTable(this, tableName)
  }

  /**
    * :: Experimental ::
    * Interface for saving the content of the [[DataFrame]] out into external storage.
    *
    * @group output
    * @since 1.4.0
    */
  @Experimental
  def write: DataFrameWriter = new DataFrameWriter(this)

  /**
    * Returns the content of the [[DataFrame]] as a RDD of JSON strings.
    *
    * @group rdd
    * @since 1.3.0
    */
  def toJSON: RDD[String] = {
    val rowSchema = this.schema
    this.mapPartitions { iter =>
      val writer = new CharArrayWriter()
      // create the Generator without separator inserted between 2 records
      val gen = new JsonFactory().createGenerator(writer).setRootValueSeparator(null)

      new Iterator[String] {
        override def hasNext: Boolean = iter.hasNext

        override def next(): String = {
          JacksonGenerator(rowSchema, gen)(iter.next())
          gen.flush()

          val json = writer.toString
          if (hasNext) {
            writer.reset()
          } else {
            gen.close()
          }

          json
        }
      }
    }
  }

  /**
    * Returns a best-effort snapshot of the files that compose this DataFrame. This method simply
    * asks each constituent BaseRelation for its respective files and takes the union of all results.
    * Depending on the source relations, this may not find all input files. Duplicates are removed.
    */
  def inputFiles: Array[String] = {
    val files: Seq[String] = logicalPlan.collect {
      case LogicalRelation(fsBasedRelation: FileRelation) =>
        fsBasedRelation.inputFiles
      case fr: FileRelation =>
        fr.inputFiles
    }.flatten
    files.toSet.toArray
  }

  ////////////////////////////////////////////////////////////////////////////
  // for Python API
  ////////////////////////////////////////////////////////////////////////////

  /**
    * Converts a JavaRDD to a PythonRDD.
    */
  protected[sql] def javaToPython: JavaRDD[Array[Byte]] = {
    val structType = schema // capture it for closure
    val rdd = queryExecution.toRdd.map(EvaluatePython.toJava(_, structType))
    EvaluatePython.javaToPython(rdd)
  }

  ////////////////////////////////////////////////////////////////////////////
  ////////////////////////////////////////////////////////////////////////////
  // Deprecated methods
  ////////////////////////////////////////////////////////////////////////////
  ////////////////////////////////////////////////////////////////////////////

  /**
    * @deprecated As of 1.3.0, replaced by `toDF()`.
    */
  @deprecated("use toDF", "1.3.0")
  def toSchemaRDD: DataFrame = this

  /**
    * Save this [[DataFrame]] to a JDBC database at `url` under the table name `table`.
    * This will run a `CREATE TABLE` and a bunch of `INSERT INTO` statements.
    * If you pass `true` for `allowExisting`, it will drop any table with the
    * given name; if you pass `false`, it will throw if the table already
    * exists.
    *
    * @group output
    * @deprecated As of 1.340, replaced by `write().jdbc()`.
    */
  @deprecated("Use write.jdbc()", "1.4.0")
  def createJDBCTable(url: String, table: String, allowExisting: Boolean): Unit = {
    val w = if (allowExisting) write.mode(SaveMode.Overwrite) else write
    w.jdbc(url, table, new Properties)
  }

  /**
    * Save this [[DataFrame]] to a JDBC database at `url` under the table name `table`.
    * Assumes the table already exists and has a compatible schema.  If you
    * pass `true` for `overwrite`, it will `TRUNCATE` the table before
    * performing the `INSERT`s.
    *
    * The table must already exist on the database.  It must have a schema
    * that is compatible with the schema of this RDD; inserting the rows of
    * the RDD in order via the simple statement
    * `INSERT INTO table VALUES (?, ?, ..., ?)` should not fail.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().jdbc()`.
    */
  @deprecated("Use write.jdbc()", "1.4.0")
  def insertIntoJDBC(url: String, table: String, overwrite: Boolean): Unit = {
    val w = if (overwrite) write.mode(SaveMode.Overwrite) else write
    w.jdbc(url, table, new Properties)
  }

  /**
    * Saves the contents of this [[DataFrame]] as a parquet file, preserving the schema.
    * Files that are written out using this method can be read back in as a [[DataFrame]]
    * using the `parquetFile` function in [[SQLContext]].
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().parquet()`.
    */
  @deprecated("Use write.parquet(path)", "1.4.0")
  def saveAsParquetFile(path: String): Unit = {
    write.format("parquet").mode(SaveMode.ErrorIfExists).save(path)
  }

  /**
    * Creates a table from the the contents of this DataFrame.
    * It will use the default data source configured by spark.sql.sources.default.
    * This will fail if the table already exists.
    *
    * Note that this currently only works with DataFrames that are created from a HiveContext as
    * there is no notion of a persisted catalog in a standard SQL context.  Instead you can write
    * an RDD out to a parquet file, and then register that file as a table.  This "table" can then
    * be the target of an `insertInto`.
    *
    * When the DataFrame is created from a non-partitioned [[HadoopFsRelation]] with a single input
    * path, and the data source provider can be mapped to an existing Hive builtin SerDe (i.e. ORC
    * and Parquet), the table is persisted in a Hive compatible format, which means other systems
    * like Hive will be able to read this table. Otherwise, the table is persisted in a Spark SQL
    * specific format.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().saveAsTable(tableName)`.
    */
  @deprecated("Use write.saveAsTable(tableName)", "1.4.0")
  def saveAsTable(tableName: String): Unit = {
    write.mode(SaveMode.ErrorIfExists).saveAsTable(tableName)
  }

  /**
    * Creates a table from the the contents of this DataFrame, using the default data source
    * configured by spark.sql.sources.default and [[SaveMode.ErrorIfExists]] as the save mode.
    *
    * Note that this currently only works with DataFrames that are created from a HiveContext as
    * there is no notion of a persisted catalog in a standard SQL context.  Instead you can write
    * an RDD out to a parquet file, and then register that file as a table.  This "table" can then
    * be the target of an `insertInto`.
    *
    * When the DataFrame is created from a non-partitioned [[HadoopFsRelation]] with a single input
    * path, and the data source provider can be mapped to an existing Hive builtin SerDe (i.e. ORC
    * and Parquet), the table is persisted in a Hive compatible format, which means other systems
    * like Hive will be able to read this table. Otherwise, the table is persisted in a Spark SQL
    * specific format.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().mode(mode).saveAsTable(tableName)`.
    */
  @deprecated("Use write.mode(mode).saveAsTable(tableName)", "1.4.0")
  def saveAsTable(tableName: String, mode: SaveMode): Unit = {
    write.mode(mode).saveAsTable(tableName)
  }

  /**
    * Creates a table at the given path from the the contents of this DataFrame
    * based on a given data source and a set of options,
    * using [[SaveMode.ErrorIfExists]] as the save mode.
    *
    * Note that this currently only works with DataFrames that are created from a HiveContext as
    * there is no notion of a persisted catalog in a standard SQL context.  Instead you can write
    * an RDD out to a parquet file, and then register that file as a table.  This "table" can then
    * be the target of an `insertInto`.
    *
    * When the DataFrame is created from a non-partitioned [[HadoopFsRelation]] with a single input
    * path, and the data source provider can be mapped to an existing Hive builtin SerDe (i.e. ORC
    * and Parquet), the table is persisted in a Hive compatible format, which means other systems
    * like Hive will be able to read this table. Otherwise, the table is persisted in a Spark SQL
    * specific format.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().format(source).saveAsTable(tableName)`.
    */
  @deprecated("Use write.format(source).saveAsTable(tableName)", "1.4.0")
  def saveAsTable(tableName: String, source: String): Unit = {
    write.format(source).saveAsTable(tableName)
  }

  /**
    * :: Experimental ::
    * Creates a table at the given path from the the contents of this DataFrame
    * based on a given data source, [[SaveMode]] specified by mode, and a set of options.
    *
    * Note that this currently only works with DataFrames that are created from a HiveContext as
    * there is no notion of a persisted catalog in a standard SQL context.  Instead you can write
    * an RDD out to a parquet file, and then register that file as a table.  This "table" can then
    * be the target of an `insertInto`.
    *
    * When the DataFrame is created from a non-partitioned [[HadoopFsRelation]] with a single input
    * path, and the data source provider can be mapped to an existing Hive builtin SerDe (i.e. ORC
    * and Parquet), the table is persisted in a Hive compatible format, which means other systems
    * like Hive will be able to read this table. Otherwise, the table is persisted in a Spark SQL
    * specific format.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().mode(mode).saveAsTable(tableName)`.
    */
  @deprecated("Use write.format(source).mode(mode).saveAsTable(tableName)", "1.4.0")
  def saveAsTable(tableName: String, source: String, mode: SaveMode): Unit = {
    write.format(source).mode(mode).saveAsTable(tableName)
  }

  /**
    * Creates a table at the given path from the the contents of this DataFrame
    * based on a given data source, [[SaveMode]] specified by mode, and a set of options.
    *
    * Note that this currently only works with DataFrames that are created from a HiveContext as
    * there is no notion of a persisted catalog in a standard SQL context.  Instead you can write
    * an RDD out to a parquet file, and then register that file as a table.  This "table" can then
    * be the target of an `insertInto`.
    *
    * When the DataFrame is created from a non-partitioned [[HadoopFsRelation]] with a single input
    * path, and the data source provider can be mapped to an existing Hive builtin SerDe (i.e. ORC
    * and Parquet), the table is persisted in a Hive compatible format, which means other systems
    * like Hive will be able to read this table. Otherwise, the table is persisted in a Spark SQL
    * specific format.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by
    *             `write().format(source).mode(mode).options(options).saveAsTable(tableName)`.
    */
  @deprecated("Use write.format(source).mode(mode).options(options).saveAsTable(tableName)",
    "1.4.0")
  def saveAsTable(
                   tableName: String,
                   source: String,
                   mode: SaveMode,
                   options: java.util.Map[String, String]): Unit = {
    write.format(source).mode(mode).options(options).saveAsTable(tableName)
  }

  /**
    * (Scala-specific)
    * Creates a table from the the contents of this DataFrame based on a given data source,
    * [[SaveMode]] specified by mode, and a set of options.
    *
    * Note that this currently only works with DataFrames that are created from a HiveContext as
    * there is no notion of a persisted catalog in a standard SQL context.  Instead you can write
    * an RDD out to a parquet file, and then register that file as a table.  This "table" can then
    * be the target of an `insertInto`.
    *
    * When the DataFrame is created from a non-partitioned [[HadoopFsRelation]] with a single input
    * path, and the data source provider can be mapped to an existing Hive builtin SerDe (i.e. ORC
    * and Parquet), the table is persisted in a Hive compatible format, which means other systems
    * like Hive will be able to read this table. Otherwise, the table is persisted in a Spark SQL
    * specific format.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by
    *             `write().format(source).mode(mode).options(options).saveAsTable(tableName)`.
    */
  @deprecated("Use write.format(source).mode(mode).options(options).saveAsTable(tableName)",
    "1.4.0")
  def saveAsTable(
                   tableName: String,
                   source: String,
                   mode: SaveMode,
                   options: Map[String, String]): Unit = {
    write.format(source).mode(mode).options(options).saveAsTable(tableName)
  }

  /**
    * Saves the contents of this DataFrame to the given path,
    * using the default data source configured by spark.sql.sources.default and
    * [[SaveMode.ErrorIfExists]] as the save mode.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().save(path)`.
    */
  @deprecated("Use write.save(path)", "1.4.0")
  def save(path: String): Unit = {
    write.save(path)
  }

  /**
    * Saves the contents of this DataFrame to the given path and [[SaveMode]] specified by mode,
    * using the default data source configured by spark.sql.sources.default.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().mode(mode).save(path)`.
    */
  @deprecated("Use write.mode(mode).save(path)", "1.4.0")
  def save(path: String, mode: SaveMode): Unit = {
    write.mode(mode).save(path)
  }

  /**
    * Saves the contents of this DataFrame to the given path based on the given data source,
    * using [[SaveMode.ErrorIfExists]] as the save mode.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().format(source).save(path)`.
    */
  @deprecated("Use write.format(source).save(path)", "1.4.0")
  def save(path: String, source: String): Unit = {
    write.format(source).save(path)
  }

  /**
    * Saves the contents of this DataFrame to the given path based on the given data source and
    * [[SaveMode]] specified by mode.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by `write().format(source).mode(mode).save(path)`.
    */
  @deprecated("Use write.format(source).mode(mode).save(path)", "1.4.0")
  def save(path: String, source: String, mode: SaveMode): Unit = {
    write.format(source).mode(mode).save(path)
  }

  /**
    * Saves the contents of this DataFrame based on the given data source,
    * [[SaveMode]] specified by mode, and a set of options.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by
    *             `write().format(source).mode(mode).options(options).save(path)`.
    */
  @deprecated("Use write.format(source).mode(mode).options(options).save()", "1.4.0")
  def save(
            source: String,
            mode: SaveMode,
            options: java.util.Map[String, String]): Unit = {
    write.format(source).mode(mode).options(options).save()
  }

  /**
    * (Scala-specific)
    * Saves the contents of this DataFrame based on the given data source,
    * [[SaveMode]] specified by mode, and a set of options
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by
    *             `write().format(source).mode(mode).options(options).save(path)`.
    */
  @deprecated("Use write.format(source).mode(mode).options(options).save()", "1.4.0")
  def save(
            source: String,
            mode: SaveMode,
            options: Map[String, String]): Unit = {
    write.format(source).mode(mode).options(options).save()
  }

  /**
    * Adds the rows from this RDD to the specified table, optionally overwriting the existing data.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by
    *             `write().mode(SaveMode.Append|SaveMode.Overwrite).saveAsTable(tableName)`.
    */
  @deprecated("Use write.mode(SaveMode.Append|SaveMode.Overwrite).saveAsTable(tableName)", "1.4.0")
  def insertInto(tableName: String, overwrite: Boolean): Unit = {
    write.mode(if (overwrite) SaveMode.Overwrite else SaveMode.Append).insertInto(tableName)
  }

  /**
    * Adds the rows from this RDD to the specified table.
    * Throws an exception if the table already exists.
    *
    * @group output
    * @deprecated As of 1.4.0, replaced by
    *             `write().mode(SaveMode.Append).saveAsTable(tableName)`.
    */
  @deprecated("Use write.mode(SaveMode.Append).saveAsTable(tableName)", "1.4.0")
  def insertInto(tableName: String): Unit = {
    write.mode(SaveMode.Append).insertInto(tableName)
  }

  /**
    * Wrap a DataFrame action to track all Spark jobs in the body so that we can connect them with
    * an execution.
    */
  private[sql] def withNewExecutionId[T](body: => T): T = {
    SQLExecution.withNewExecutionId(sqlContext, queryExecution)(body)
  }

  ////////////////////////////////////////////////////////////////////////////
  ////////////////////////////////////////////////////////////////////////////
  // End of deprecated methods
  ////////////////////////////////////////////////////////////////////////////
  ////////////////////////////////////////////////////////////////////////////

}

redpajama_set_name: RedPajamaGithub


SPHK1 and SPHK2 are encoded by different genes with some redundancy of function; genetic deletion of both Sphk1 and Sphk2, but not either alone, is embryonic lethal in mice. There are splice variants of each isoform (SphK1a-c and SphK2a, b), distinguished by their N-terminal sequences. SPHK1 and SPHK2 differ in tissue distribution, sub-cellular localisation, biochemical properties and regulation. They regulate discrete pools of S1P. Receptor stimulation induces SPHK1 translocation from the cytoplasm to the plasma membrane. SPHK1 translocation is regulated by phosphorylation/dephosphorylation, specific protein:protein interactions and interaction with specific lipids at the plasma membrane. SPHK1 is a dimeric protein, as confirmed by its crystal structure which forms a positive cluster, between protomers, essential for interaction with anionic phospholipids in the plasma membrane. SPHK2 is localised to the ER or associated with mitochondria or shuttles in/out of the nucleus, regulated by phosphorylation. Intracellular targets of nuclear S1P include the catalytic subunit of telomerase (TERT) and regulators of gene expression including histone deacetylases (HDAC 1/2) and peroxisome proliferator-activated receptor gamma (PPARγ). SPHK2 phosphorylates the pro-drug FTY720 (fingolimod, which is used to treat some forms of multiple sclerosis) to a mimic of S1P and that acts as a functional antagonist of S1P1 receptors. Inhibitors of SPHK1 and SPHK2 have therapeutic potential in many diseases.
Comment SK1 inhibitors induce its proteasomal degradation [6-7]. SK1 crystal structures confirm that it is dimeric ; there is no crystal structure available for SK2.
Comment There is no crystal structure available for SK2.
MP-A08 is competitive with ATP; other SPHK inhibitors are competitive with sphingosine. ABC294640 (opaganib) has known off-target effects on dihydroceramide desaturase (DEGS1) [7,11]) and induces proteasomal degradation of SK1 . ABC294640 is in clinical trials for advanced cholangiocarcinoma, advanced hepatocellular carcinoma and refractory/relapsed multiple myeloma (to view ClinicalTrials.gov list click here).
Nigel J Pyne, Susan Pyne. Sphingosine kinase. Accessed on 25/04/2019. IUPHAR/BPS Guide to PHARMACOLOGY, http://www.guidetopharmacology.org/GRAC/FamilyDisplayForward?familyId=787.


Henk Helmantel (1945) is a true master of the fine arts and is generally seen as one of the most important figures within the movement of Contemporary Realism. His famous still lifes are admired by the viewer for their air of quiet serenity. Less well known is the fact that he paints magnificent church interiors and can sometimes be persuaded to depict a rare landscape. Even a couple of self-portraits can be found in his oeuvre. Helmantel uses a set of sober colours such as grey, green and blue. Specific details are emphasized with the use of more vivid hues. The artist finds inspiration in the works of his Dutch Golden Age predecessors. The dramatic highlights used by Rembrandt van Rijn are to be found in Helmantels paintings. And the soft and balanced nuance of colours are an homage to Johannes Vermeer.
Real beauty can be found all around us, according to Henk Helmantel. Even in the objects of everyday life: a round, shiny plum or the deep red of an apple. Beauty can be seen in an old wooden table that has seen years of use, or in the faded green of an old glass bottle. Helmantel's work is a quiet contemplation on the serenity of creation. With his artworks, Helmantel invites the viewer to re-examine one's surroundings in search of this everyday beauty.
Helmantel's paintings are all about finding and combining the right objects. Even the plainest will do: a round piece of bread, a bottle in a vivid green shade or an ancient metal pitcher. These paraphernalia are arranged with the utmost care on a table by the window. The most important factor now to be decided upon is the composition of the objects. A balanced composition consists of (at least) one horizontal, one vertical and, one diagonal line. These lines together must form a balanced whole. The artist places and replaces objects until he is satisfied with the arrangement. Only then is he ready to start working on the actual paintings. He starts with a drawing directly on the panel, after which he picks up his brush. With each thin layer of oil paint, the image gains clarity, detail and depth of colour. The objects gain solidity and substance. The final phase is for crossing the t's and dotting the i's: the amount of light is adjusted, the last details are filled in.
A large collection of paintings by Henk Helmantel is for sale at Collection Harms. In addition, Helmantel's oeuvre has been shown in several leading museums, amongst which the Drents Museum and the Groninger Museum. His works can be found all over the globe: they have can be found in Germany, Switzerland, Taiwan and the U.S., amongst many other countries.
Exhibition on Henk Helmantel in Museum Gouda.


Dallas Mayor Pro Tem Dwaine Caraway, the second most powerful official at city hall, is going to federal prison.
Caraway pleaded guilty to criminal conspiracy, wire fraud and tax evasion.
The "appropriate " punishment for Caraway is one "that does not exceed seven years, or 84 months, imprisonment," documents from the U.S. Attorney's Office said.
NBC 5 will be live-streaming here @ the 10 a.m. news conference at the Dallas office of the U.S. Attorney.


Thurston County Superior Court Judge James Dixon this morning granted a request from the National Rifle Association (NRA) and right wing activist Alan Gottlieb to block the Alliance for Gun Responsibility's Initiative 1639 from the November 2018 statewide ballot, deeming I-1639's petitions to be fatally defective.
It's the first time in state history that an initiative certified for a statewide vote by the Secretary of State has been ordered off the ballot. However, Judge Dixon's ruling will not be the final word on I-1639's fate, as the campaign has already appealed his decision to the Washington State Supreme Court.
I-1639 seeks to raise the minimum age required to purchase semiautomatic firearms, impose new safe storage requirements, and create an enhanced background check system. Its signature drive took place over the span of only four weeks, owing to the Alliance waiting until after the legislative session had adjourned Sine Die to decide to go to the ballot and launch a campaign.
When I-1639's petitions were prepared, an error resulted in the text on the reverse side of the petitions appearing without the usual markup (underlines and strikethroughs) commonly used in legal documents to signify new text and deleted text. The NRA and Gottlieb have argued that this defect should result in all of the petitions being invalidated — an argument Dixon found persuasive.
The campaign has countered that invalidation of the petitions would cause hundreds of thousands of Washington voters to be disenfranchised.
In 2015, when progressive organizations sued to block Tim Eyman's I-1366 from the ballot on scope grounds in Huff v. Wyman, Eyman and his amen chorus cried foul, asserting that the First Amendment rights of the voters who signed the measure would be violated if I-1366 was voided in a preelection challenge.
Yesterday, however, Eyman publicly called on Judge Dixon to nullify I-1639 in a statement intended to show solidarity with the NRA and Gottlieb.
Wow! That is some incredibly tortured logic right there.
No one is harmed by a public vote on an initiative. Tell that to the NRA and Alan Gottlieb, who brought this legal challenge against I-1639. They're clearly afraid of being handed their third consecutive ballot defeat in November, and they don't want the public to have the opportunity to decide the fate of this initiative.
The NRA and Gottlieb's don't-let-the-people vote lawsuit was to be expected. What is more surprising is that Tim Eyman is publicly supporting it.
Eyman not only released a statement cheering on the NRA and Gottlieb and calling for I-1639's demise, he went down to Thurston County Superior Court for the hearing in front of Judge Dixon and afterward recorded a video message for his followers from his car, which he posted on social media.
Regardless of what defects the I-1639 petitions have, the First Amendment to the United States Constitution cannot apply to some and not others.
If blocking I-1366 from the ballot in 2015 would have violated the United States Constitution (as Eyman and his attorneys argued then), then blocking I-1639 from the ballot this year would also violate the United States Constitution.
The right wing cannot have it both ways.
The courts have never stopped a statewide initiative that had enough signatures to qualify for a spot on the ballot from going to voters. In siding with the NRA and Gottlieb, Judge Dixon has just taken the "unprecedented and undemocratic step of preventing the people from voting on a qualified statewide initiative" that former State Senator Pam Roach warned against in 2015.
But again, Dixon won't have the final word on I-1639. The Washington State Supreme Court will. The state's highest court normally takes months to hear and decide a case like this, but since ballots have to be designed and mailed within a very short timeframe, the Court will have to decide within days.
Contrary to what Tim Eyman told his followers in his video message, the Court will be accepting briefs from the parties in this matter. It may also hear oral argument.
"Both sides anticipated that the loser would appeal to the state Supreme Court, and with a deadline for the printing of ballots and the state Voter's Pamphlet looming, an expedited schedule has already been set," the Spokesman-Review's Olympia correspondent Jim Camden explained to readers in his report on Dixon's ruling.
If the Supreme Court sticks to precedent, then Dixon's ruling will be overturned and voters will get the opportunity to vote on I-1639 this fall. Otherwise, we could end up with a new precedent that will have serious ramifications for future initiatives.
Regardless of Tim Eyman being a waste of oxygen, the campaign to get I-1639 on the ballot was pretty shady. Oddly enough, I saw at least one signature gatherer collecting signatures for both I-1639 *and* whatever Eyman's latest car tab nonsense was.
© 2019 The Northwest Progressive Institute. The Cascadia Advocate loaded in 0.197 seconds.


Service Experts Heating and Air Conditioning surveyed 1,000 homeowners in the U.S. and Canada on their understanding of carbon monoxide (CO). According to The Centers for Disease Control, carbon monoxide poisoning is a leading cause of unintentional poisoning deaths – causing approximately 15,000 emergency room visits and almost 500 deaths every year in the United States1. There are alarming gaps in what homeowners do and don't know about remaining safe.
Carbon monoxide is a colorless, odorless gas that may build to harmful levels in your home when fuel-burning devices are not correctly vented, operated or maintained. Breathing in CO can cause carbon monoxide poisoning which attacks the nervous system, heart and could be fatal. It's critical to have your home inspected for carbon monoxide regularly to help ensure safe air quality.
More than half of homeowners (54%) don't know that the clothes dryer could be a source of carbon monoxide; 25% don't know that that fireplace also can be a source. Other likely sources include gas water heaters, gas ranges, charcoal grills, gas space heaters and wood burning stoves.
Over one third of homeowners (34%) don't have or aren't sure if they have a carbon monoxide detector in their home; 41% say they never change their carbon monoxide detector, or they aren't sure how often it is replaced.
With a simple home comfort analysis, Service Experts Heating & Air Conditioning in West Palm Beach is happy to help keep your family safe this winter. Visit ServiceExperts.com/CO-Safety, call 561-629-1826 or use our online scheduler to set up an analysis of CO and other harmful indoor air pollutant levels.