import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

SparkSession spark = ...   // create SparkSession

Dataset<Row> data = data = spark.range(0, 5);
data.write().format("delta").save("/tmp/delta-table");

test("append then read") {
  val tempDir = Utils.createTempDir()
  Seq(1).toDF().write.format("delta").save(tempDir.toString)
  Seq(2, 3).toDF().write.format("delta").mode("append").save(tempDir.toString)

  def data: DataFrame = spark.read.format("delta").load(tempDir.toString)
  checkAnswer(data, Row(1) :: Row(2) :: Row(3) :: Nil)

  // append more
  Seq(4, 5, 6).toDF().write.format("delta").mode("append").save(tempDir.toString)
  checkAnswer(data.toDF(), Row(1) :: Row(2) :: Row(3) :: Row(4) :: Row(5) :: Row(6) :: Nil)
}

test("append mode") {
  failAfter(streamingTimeout) {
    withTempDirs { (outputDir, checkpointDir) =>
      val inputData = MemoryStream[Int]
      val df = inputData.toDF()
      val query = df.writeStream
        .option("checkpointLocation", checkpointDir.getCanonicalPath)
        .format("delta")
        .start(outputDir.getCanonicalPath)
      val log = DeltaLog.forTable(spark, outputDir.getCanonicalPath)
      try {
        inputData.addData(1)
        query.processAllAvailable()

        val outputDf = spark.read.format("delta").load(outputDir.getCanonicalPath)
        checkDatasetUnorderly(outputDf.as[Int], 1)
        assert(log.update().transactions.head == (query.id.toString -> 0L))

        inputData.addData(2)
        query.processAllAvailable()

        checkDatasetUnorderly(outputDf.as[Int], 1, 2)
        assert(log.update().transactions.head == (query.id.toString -> 1L))

        inputData.addData(3)
        query.processAllAvailable()

        checkDatasetUnorderly(outputDf.as[Int], 1, 2, 3)
        assert(log.update().transactions.head == (query.id.toString -> 2L))
      } finally {
        query.stop()
      }
    }
  }
}

val data = spark.range(0, 5)
data.write.format("delta").save("/tmp/delta-table")

val path = parameters.getOrElse("path", {
  throw DeltaErrors.pathNotSpecifiedException
})
val partitionColumns = parameters.get(DeltaSourceUtils.PARTITIONING_COLUMNS_KEY)
  .map(DeltaDataSource.decodePartitioningColumns)
  .getOrElse(Nil)

val deltaLog = DeltaLog.forTable(sqlContext.sparkSession, path)

override def run(sparkSession: SparkSession): Seq[Row] = {
  deltaLog.withNewTransaction { txn =>
    val actions = write(txn, sparkSession)
    val operation = DeltaOperations.Write(mode, Option(partitionColumns), options.replaceWhere)
    txn.commit(actions, operation)
  }
  Seq.empty
}

updateMetadata(txn, data, partitionColumns, configuration, isOverwriteOperation, rearrangeOnly)

val replaceWhere = options.replaceWhere
val partitionFilters = if (replaceWhere.isDefined) {
  val predicates = parsePartitionPredicates(sparkSession, replaceWhere.get)
  if (mode == SaveMode.Overwrite) {
    verifyPartitionPredicates(
      sparkSession, txn.metadata.partitionColumns, predicates)
  }
  Some(predicates)
} else {
  None
}

val newFiles = txn.writeFiles(data, Some(options))

val operation = DeltaOperations.Write(mode, Option(partitionColumns), options.replaceWhere)

txn.commit(actions, operation)

val df = spark.read.format("delta").load("/tmp/delta-table")
df.show()

val df = spark.read.format("delta").load("/tmp/delta-table")

val maybeTimeTravel =
  DeltaTableUtils.extractIfPathContainsTimeTravel(sqlContext.sparkSession, maybePath)

val hadoopPath = new Path(path)
val rootPath = DeltaTableUtils.findDeltaTableRoot(sqlContext.sparkSession, hadoopPath)
  .getOrElse {
    val fs = hadoopPath.getFileSystem(sqlContext.sparkSession.sessionState.newHadoopConf())
    if (!fs.exists(hadoopPath)) {
      throw DeltaErrors.pathNotExistsException(path)
    }
    hadoopPath
  }

val deltaLog = DeltaLog.forTable(sqlContext.sparkSession, rootPath)

    val partitionFilters = if (rootPath != hadoopPath) {

(snip)

      if (files.count() == 0) {
        throw DeltaErrors.pathNotExistsException(path)
      }
      filters
    } else {
      Nil
    }

deltaLog.createRelation(partitionFilters, timeTravelByParams.orElse(timeTravelByPath))

  def createRelation(
      partitionFilters: Seq[Expression] = Nil,
      timeTravel: Option[DeltaTimeTravelSpec] = None): BaseRelation = {

(snip)

    new HadoopFsRelation(
      fileIndex,
      partitionSchema = snapshotToUse.metadata.partitionSchema,
      dataSchema = snapshotToUse.metadata.schema,
      bucketSpec = None,
      snapshotToUse.fileFormat,
      snapshotToUse.metadata.format.options)(spark) with InsertableRelation {
      def insert(data: DataFrame, overwrite: Boolean): Unit = {
        val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Append
        WriteIntoDelta(
          deltaLog = DeltaLog.this,
          mode = mode,
          new DeltaOptions(Map.empty[String, String], spark.sessionState.conf),
          partitionColumns = Seq.empty,
          configuration = Map.empty,
          data = data).run(spark)
      }
    }
  }
}

abstract class TahoeFileIndex(
    val spark: SparkSession,
    val deltaLog: DeltaLog,
    val path: Path) extends FileIndex {

scala> df.show()

matchingFiles(partitionFilters, dataFilters).groupBy(_.partitionValues).map {

override def matchingFiles(
    partitionFilters: Seq[Expression],
    dataFilters: Seq[Expression],
    keepStats: Boolean = false): Seq[AddFile] = {
  getSnapshot(stalenessAcceptable = false).filesForScan(
    projection = Nil, this.partitionFilters ++ partitionFilters ++ dataFilters, keepStats).files
}

case (partitionValues, files) =>
  val rowValues: Array[Any] = partitionSchema.map { p =>
    Cast(Literal(partitionValues(p.name)), p.dataType, Option(timeZone)).eval()
  }.toArray


  val fileStats = files.map { f =>
    new FileStatus(
      /* length */ f.size,
      /* isDir */ false,
      /* blockReplication */ 0,
      /* blockSize */ 1,
      /* modificationTime */ f.modificationTime,
      absolutePath(f.path))
  }.toArray

  PartitionDirectory(new GenericInternalRow(rowValues), fileStats)

def show(numRows: Int, truncate: Boolean): Unit = if (truncate) {
  println(showString(numRows, truncate = 20))
} else {
  println(showString(numRows, truncate = 0))
}

  private[sql] def showString(
      _numRows: Int,
      truncate: Int = 20,
      vertical: Boolean = false): String = {
    val numRows = _numRows.max(0).min(ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH - 1)
    // Get rows represented by Seq[Seq[String]], we may get one more line if it has more data.
    val tmpRows = getRows(numRows, truncate)

(snip)

  private[sql] def getRows(

(snip)

    val data = newDf.select(castCols: _*).take(numRows + 1)
  
(snip)

def take(n: Int): Array[T] = head(n)

def head(n: Int): Array[T] = withAction("head", limit(n).queryExecution)(collectFromPlan)

private def collectFromPlan(plan: SparkPlan): Array[T] = {
  // This projection writes output to a `InternalRow`, which means applying this projection is not
  // thread-safe. Here we create the projection inside this method to make `Dataset` thread-safe.
  val objProj = GenerateSafeProjection.generate(deserializer :: Nil)
  plan.executeCollect().map { row =>
    // The row returned by SafeProjection is `SpecificInternalRow`, which ignore the data type
    // parameter of its `get` method, so it's safe to use null here.
    objProj(row).get(0, null).asInstanceOf[T]
  }
}

case class CollectLimitExec(limit: Int, child: SparkPlan) extends UnaryExecNode {
  override def output: Seq[Attribute] = child.output
  override def outputPartitioning: Partitioning = SinglePartition
  override def executeCollect(): Array[InternalRow] = child.executeTake(limit)

(snip)

def executeTake(n: Int): Array[InternalRow] = {
  if (n == 0) {
    return new Array[InternalRow](0)
  }

  val childRDD = getByteArrayRdd(n).map(_._2)

  private def getByteArrayRdd(n: Int = -1): RDD[(Long, Array[Byte])] = {
    execute().mapPartitionsInternal { iter =>
      var count = 0

(snip)

final def execute(): RDD[InternalRow] = executeQuery {
  if (isCanonicalizedPlan) {
    throw new IllegalStateException("A canonicalized plan is not supposed to be executed.")
  }
  doExecute()
}

protected final def executeQuery[T](query: => T): T = {
  RDDOperationScope.withScope(sparkContext, nodeName, false, true) {
    prepare()
    waitForSubqueries()
    query
  }
}

  override def doExecute(): RDD[InternalRow] = {

(snip)

    val durationMs = longMetric("pipelineTime")

    val rdds = child.asInstanceOf[CodegenSupport].inputRDDs()
    assert(rdds.size <= 2, "Up to two input RDDs can be supported")

(snip)

override def inputRDDs(): Seq[RDD[InternalRow]] = {
  inputRDD :: Nil
}

private lazy val inputRDD: RDD[InternalRow] = {
  // Update metrics for taking effect in both code generation node and normal node.
  updateDriverMetrics()
  val readFile: (PartitionedFile) => Iterator[InternalRow] =
    relation.fileFormat.buildReaderWithPartitionValues(
      sparkSession = relation.sparkSession,
      dataSchema = relation.dataSchema,
      partitionSchema = relation.partitionSchema,
      requiredSchema = requiredSchema,
      filters = pushedDownFilters,
      options = relation.options,
      hadoopConf = relation.sparkSession.sessionState.newHadoopConfWithOptions(relation.options))

  relation.bucketSpec match {
    case Some(bucketing) if relation.sparkSession.sessionState.conf.bucketingEnabled =>
      createBucketedReadRDD(bucketing, readFile, selectedPartitions, relation)
    case _ =>
      createNonBucketedReadRDD(readFile, selectedPartitions, relation)
  }
}

val data = spark.range(5, 10)
data.write.format("delta").mode("overwrite").save("/tmp/delta-table")

data.write.format("delta").mode("overwrite").save("/tmp/delta-table")

override def run(sparkSession: SparkSession): Seq[Row] = {
  deltaLog.withNewTransaction { txn =>
    val actions = write(txn, sparkSession)
    val operation = DeltaOperations.Write(mode, Option(partitionColumns), options.replaceWhere)
    txn.commit(actions, operation)
  }
  Seq.empty
}

scala> val streamingDf = spark.readStream.format("rate").load()
scala> val stream = streamingDf.select($"value" as "id").writeStream.format("delta").option("checkpointLocation", "/tmp/checkpoint").start("/tmp/delta-table")

scala> val df = spark.read.format("delta").load("/tmp/delta-table")

scala> df.count
res1: Long = 1139

scala> df.count
res2: Long = 1158

scala> df.groupBy("id").count.sort($"count".desc).show
+----+-----+
|  id|count|
+----+-----+
| 964|    1|
|  29|    1|
|  26|    1|
| 474|    1|
| 831|    1|
|1042|    1|
| 167|    1|
| 112|    1|
| 299|    1|
| 155|    1|
| 385|    1|
| 736|    1|
| 113|    1|
|1055|    1|
| 502|    1|
|1480|    1|
| 656|    1|
| 287|    1|
|   0|    1|
| 348|    1|
+----+-----+

scala> stream.stop()

scala> val stream = streamingDf.select($"value" % 10 as "id").writeStream.format("delta").option("checkpointLocation", "/tmp/checkpoint").start("/tmp/delta-table")

scala> df.groupBy("id").count.sort($"count".desc).show
+----+-----+
|  id|count|
+----+-----+
|   0|   19|
|   9|   19|
|   1|   19|
|   2|   19|
|   3|   18|
|   6|   18|
|   5|   18|
|   8|   18|
|   7|   18|
|   4|   18|
|1055|    1|
|1547|    1|
|  26|    1|
|1224|    1|
| 502|    1|
|1480|    1|
| 237|    1|
| 588|    1|
| 602|    1|
| 347|    1|
+----+-----+
only showing top 20 rows

scala> val stream2 = spark.readStream.format("delta").load("/tmp/delta-table").writeStream.format("console").start()

scala> stream.stop()

scala> stream2.stop()

scala> val stream = streamingDf.select($"value" as "id").writeStream.format("delta").option("checkpointLocation", "/tmp/checkpoint").start("/tmp/delta-table")

override def createSink(
    sqlContext: SQLContext,
    parameters: Map[String, String],
    partitionColumns: Seq[String],
    outputMode: OutputMode): Sink = {
  val path = parameters.getOrElse("path", {
    throw DeltaErrors.pathNotSpecifiedException
  })
  if (outputMode != OutputMode.Append && outputMode != OutputMode.Complete) {
    throw DeltaErrors.outputModeNotSupportedException(getClass.getName, outputMode)
  }
  val deltaOptions = new DeltaOptions(parameters, sqlContext.sparkSession.sessionState.conf)
  new DeltaSink(sqlContext, new Path(path), partitionColumns, outputMode, deltaOptions)
}

  override def addBatch(batchId: Long, data: DataFrame): Unit = deltaLog.withNewTransaction { txn =>
    val queryId = sqlContext.sparkContext.getLocalProperty(StreamExecution.QUERY_ID_KEY)
    assert(queryId != null)

    if (SchemaUtils.typeExistsRecursively(data.schema)(_.isInstanceOf[NullType])) {
      throw DeltaErrors.streamWriteNullTypeException
    }

(snip)

val selfScan = data.queryExecution.analyzed.collectFirst {
  case DeltaTable(index) if index.deltaLog.isSameLogAs(txn.deltaLog) => true
}.nonEmpty
if (selfScan) {
  txn.readWholeTable()
}

updateMetadata(
  txn,
  data,
  partitionColumns,
  configuration = Map.empty,
  outputMode == OutputMode.Complete())

val currentVersion = txn.txnVersion(queryId)
if (currentVersion >= batchId) {
  logInfo(s"Skipping already complete epoch $batchId, in query $queryId")
  return
}

val deletedFiles = outputMode match {
  case o if o == OutputMode.Complete() =>
    deltaLog.assertRemovable()
    txn.filterFiles().map(_.remove)
  case _ => Nil
}

val newFiles = txn.writeFiles(data, Some(options))

val setTxn = SetTransaction(queryId, batchId, Some(deltaLog.clock.getTimeMillis())) :: Nil
val info = DeltaOperations.StreamingUpdate(outputMode, queryId, batchId)
txn.commit(setTxn ++ newFiles ++ deletedFiles, info)

override def getBatch(start: Option[Offset], end: Offset): DataFrame = {

(snip)

val endOffset = DeltaSourceOffset(tableId, end)
previousOffset = endOffset // For recovery

val changes = if (start.isEmpty) {
  if (endOffset.isStartingVersion) {
    getChanges(endOffset.reservoirVersion, -1L, isStartingVersion = true)
  } else {
    assert(endOffset.reservoirVersion > 0, s"invalid reservoirVersion in endOffset: $endOffset")
    // Load from snapshot `endOffset.reservoirVersion - 1L` so that `index` in `endOffset`
    // is still valid.
    getChanges(endOffset.reservoirVersion - 1L, -1L, isStartingVersion = true)
  }
} else {
  val startOffset = DeltaSourceOffset(tableId, start.get)
  if (!startOffset.isStartingVersion) {
    // unpersist `snapshot` because it won't be used any more.
    cleanUpSnapshotResources()
  }
  getChanges(startOffset.reservoirVersion, startOffset.index, startOffset.isStartingVersion)
}

} else {
  val startOffset = DeltaSourceOffset(tableId, start.get)
  if (!startOffset.isStartingVersion) {
    // unpersist `snapshot` because it won't be used any more.
    cleanUpSnapshotResources()
  }
  getChanges(startOffset.reservoirVersion, startOffset.index, startOffset.isStartingVersion)
}

val addFilesIter = changes.takeWhile { case IndexedFile(version, index, _, _) =>
  version < endOffset.reservoirVersion ||
    (version == endOffset.reservoirVersion && index <= endOffset.index)
}.collect { case i: IndexedFile if i.add != null => i.add }
val addFiles =
  addFilesIter.filter(a => excludeRegex.forall(_.findFirstIn(a.path).isEmpty)).toSeq

logDebug(s"start: $start end: $end ${addFiles.toList}")
deltaLog.createDataFrame(deltaLog.snapshot, addFiles, isStreaming = true)

reportTimeTaken("addBatch") {
  SQLExecution.withNewExecutionId(sparkSessionToRunBatch, lastExecution) {
    sink match {
      case s: Sink => s.addBatch(currentBatchId, nextBatch)
      case _: StreamWriteSupport =>
        // This doesn't accumulate any data - it just forces execution of the microbatch writer.
        nextBatch.collect()
    }
  }
}

val nextBatch =
  new Dataset(sparkSessionToRunBatch, lastExecution, RowEncoder(lastExecution.analyzed.schema))

scala> // テーブルの準備
scala> val id = spark.range(0, 100000)
scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> data.write.format("delta").save("/tmp/delta-table")
scala> // いったん別のDFとして読み込み定義
scala> val df = spark.read.format("delta").load("/tmp/delta-table")
scala> // 新しいカラム付きのレコードを定義し、書き込み
scala> val newRecords = id.select($"id", rand() * 10 % 10 as "rand" cast "int", rand() * 10 % 10 as "tmp" cast "int")
scala> newRecords.write.format("delta").mode("append").save("/tmp/delta-table")

def addSchemaMismatch(original: StructType, data: StructType): Unit = {
  bits ++=
    s"""A schema mismatch detected when writing to the Delta table.
       |To enable schema migration, please set:
       |'.option("${DeltaOptions.MERGE_SCHEMA_OPTION}", "true")'.
       |
       |Table schema:
       |${DeltaErrors.formatSchema(original)}
       |
       |Data schema:
       |${DeltaErrors.formatSchema(data)}
       """.stripMargin :: Nil
  mentionedOption = true
}

if (isNewSchema) {
  errorBuilder.addSchemaMismatch(txn.metadata.schema, dataSchema)
}

val dataSchema = data.schema.asNullable
val mergedSchema = if (isOverwriteMode && canOverwriteSchema) {
  dataSchema
} else {
  SchemaUtils.mergeSchemas(txn.metadata.schema, dataSchema)
}
val normalizedPartitionCols =
  normalizePartitionColumns(data.sparkSession, partitionColumns, dataSchema)
// Merged schema will contain additional columns at the end
def isNewSchema: Boolean = txn.metadata.schema != mergedSchema

scala> // テーブルの準備
scala> val id = spark.range(0, 100000)
scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> data.write.format("delta").save("/tmp/delta-table")
scala> // いったん別のDFとして読み込み定義
scala> val df = spark.read.format("delta").load("/tmp/delta-table")
scala> // 新しいカラム付きのレコードを定義し、書き込み
scala> val newRecords = id.select($"id", rand() * 10 % 10 as "rand" cast "int", rand() * 10 % 10 as "tmp" cast "int")
scala> newRecords.write.format("delta").mode("append").option("mergeSchema", "true").save("/tmp/delta-table")

val dataSchema = data.schema.asNullable
val mergedSchema = if (isOverwriteMode && canOverwriteSchema) {
  dataSchema
} else {
  SchemaUtils.mergeSchemas(txn.metadata.schema, dataSchema)
}

def isNewSchema: Boolean = txn.metadata.schema != mergedSchema

} else if (isNewSchema && canMergeSchema && !isNewPartitioning) {
  logInfo(s"New merged schema: ${mergedSchema.treeString}")
  recordDeltaEvent(txn.deltaLog, "delta.ddl.mergeSchema")
  if (rearrangeOnly) {
    throw DeltaErrors.unexpectedDataChangeException("Change the Delta table schema")
  }
  txn.updateMetadata(txn.metadata.copy(schemaString = mergedSchema.json))

scala> // テーブルの準備
scala> val id = spark.range(0, 100000)
scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> data.write.format("delta").save("/tmp/delta-table")
scala> // いったん別のDFとして読み込み定義
scala> val df = spark.read.format("delta").load("/tmp/delta-table")
scala> // 新しいカラム付きのレコードを定義し、書き込み
scala> val newRecords = id.select($"id", rand() * 10 % 10 as "rand" cast "int", rand() * 10 % 10 as "tmp" cast "int")
scala> newRecords.write.format("delta").mode("overwrite").save("/tmp/delta-table")

scala> // テーブルの準備
scala> val id = spark.range(0, 100000)
scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> data.write.format("delta").save("/tmp/delta-table")
scala> // いったん別のDFとして読み込み定義
scala> val df = spark.read.format("delta").load("/tmp/delta-table")
scala> // 新しいカラム付きのレコードを定義し、書き込み
scala> val newRecords = id.select($"id", rand() * 10 % 10 as "rand" cast "int", rand() * 10 % 10 as "tmp" cast "int")
scala> newRecords.write.format("delta").mode("overwrite").option("overwriteSchema", "true").save("/tmp/delta-table")

} else if (isOverwriteMode && canOverwriteSchema && (isNewSchema || isNewPartitioning)) {
  // Can define new partitioning in overwrite mode
  val newMetadata = txn.metadata.copy(
    schemaString = dataSchema.json,
    partitionColumns = normalizedPartitionCols
  )
  recordDeltaEvent(txn.deltaLog, "delta.ddl.overwriteSchema")
  if (rearrangeOnly) {
    throw DeltaErrors.unexpectedDataChangeException("Overwrite the Delta table schema or " +
      "change the partition schema")
  }
  txn.updateMetadata(newMetadata)

scala> df.write.format("delta").partitionBy("date").save("/delta/events")

scala> val id = spark.range(0, 100000)
id: org.apache.spark.sql.Dataset[Long] = [id: bigint]

scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
data: org.apache.spark.sql.DataFrame = [id: bigint, rand: double]

scala> data.write.format("delta").save("/tmp/delta-table")

scala> data.write.format("delta").partitionBy("rand").save("/tmp/delta-table-partitioned")

import io.delta.tables._
import org.apache.spark.sql.functions._

val deltaTable = DeltaTable.forPath("/tmp/delta-table")

// Update every even value by adding 100 to it
deltaTable.update(
  condition = expr("id % 2 == 0"),
  set = Map("id" -> expr("id + 100")))

// Delete every even value
deltaTable.delete(condition = expr("id % 2 == 0"))

// Upsert (merge) new data
val newData = spark.range(0, 20).toDF

deltaTable.as("oldData")
  .merge(
    newData.as("newData"),
    "oldData.id = newData.id")
  .whenMatched
  .update(Map("id" -> col("newData.id")))
  .whenNotMatched
  .insert(Map("id" -> col("newData.id")))
  .execute()

deltaTable.toDF.show()

def forPath(sparkSession: SparkSession, path: String): DeltaTable = {
  if (DeltaTableUtils.isDeltaTable(sparkSession, new Path(path))) {
    new DeltaTable(sparkSession.read.format("delta").load(path),
      DeltaLog.forTable(sparkSession, path))
  } else {
    throw DeltaErrors.notADeltaTableException(DeltaTableIdentifier(path = Some(path)))
  }
}

scala> // データ準備
scala> val id = spark.range(0, 100000)
scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> data.write.format("delta").save("/tmp/delta-table")
scala> // テーブルとして読み込み定義
scala> import io.delta.tables._
scala> import org.apache.spark.sql.functions._
scala> val deltaTable = DeltaTable.forPath(spark, "/tmp/delta-table")
scala> // 条件に基づきレコードを削除
scala> deltaTable.delete("rand > 5")
scala> deltaTable.delete($"rand" > 6)
scala> deltaTable.toDF.show
+-----+----+
|   id|rand|
+-----+----+
|62500|   0|
|62501|   5|
|62503|   4|
|62504|   3|
|62505|   3|
|62506|   3|
|62507|   2|

(snip)

scala> // データ準備
scala> val id = spark.range(0, 100000)
scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> data.write.format("delta").save("/tmp/delta-table")
scala> // テーブルとして読み込み定義
scala> import io.delta.tables._
scala> import org.apache.spark.sql.functions._
scala> val deltaTable = DeltaTable.forPath(spark, "/tmp/delta-table")
scala> // 条件に基づきレコードを更新
scala> deltaTable.updateExpr("rand = 0", Map("rand" -> "-1"))

scala> deltaTable.toDF.show
+---+----+
| id|rand|
+---+----+
|  0|   4|
|  1|   6|
|  2|   0|
|  3|   5|
|  4|   1|
|  5|   6|
|  6|   7|
|  7|   0|
|  8|   0|
|  9|   5|

(snip)

scala> deltaTable.toDF.show
+---+----+
| id|rand|
+---+----+
|  0|   4|
|  1|   6|
|  2|  -1|
|  3|   5|
|  4|   1|
|  5|   6|
|  6|   7|
|  7|  -1|
|  8|  -1|

(snip)

scala> // データ準備
scala> val id = spark.range(0, 100000)
scala> val data = id.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> data.write.format("delta").save("/tmp/delta-table")
scala> // テーブルとして読み込み定義
scala> import io.delta.tables._
scala> import org.apache.spark.sql.functions._
scala> val deltaTable = DeltaTable.forPath(spark, "/tmp/delta-table")
scala> // upsert用のデータ作成
scala> val id2 = spark.range(0, 200000)
scala> val data2 = id2.select($"id", rand() * 10 % 10 as "rand" cast "int")
scala> val dataForUpsert = data2.sample(false, 0.5)
scala> // データをマージ
scala> :paste
deltaTable
  .as("before")
  .merge(
    dataForUpsert.as("updates"),
    "before.id = updates.id")
  .whenMatched
  .updateExpr(
    Map("rand" -> "updates.rand"))
  .whenNotMatched
  .insertExpr(
    Map(
      "id" -> "updates.id",
      "rand" -> "updates.rand"))
  .execute()

> deltaTable.toDF.filter($"id" > 99997).show
+-----+----+
|   id|rand|
+-----+----+
|99998|   4|
|99999|   0|
+-----+----+

scala> deltaTable.toDF.show
+---+----+
| id|rand|
+---+----+
|  0|   1|
|  1|   3|
|  2|   5|
|  3|   8|
|  4|   3|
|  5|   0|
|  6|   5|
|  7|   6|
|  8|   4|
|  9|   9|
| 10|   4|
| 11|   4|

(snip)

scala> dataForUpsert.show
+---+----+
| id|rand|
+---+----+
|  0|   7|
|  2|   6|
|  5|   2|
|  8|   9|
| 10|   9|

(snip)

scala> dataForUpsert.filter($"id" > 99997).show
+------+----+
|    id|rand|
+------+----+
| 99999|   0|
|100005|   3|
|100006|   9|
|100008|   7|
|100009|   1|

(snip)

scala> deltaTable.toDF.orderBy($"id").show
+---+----+
| id|rand|
+---+----+
|  0|   7|
|  1|   3|
|  2|   6|
|  3|   8|
|  4|   3|
|  5|   2|
|  6|   5|
|  7|   6|
|  8|   9|

(snip)

scala> deltaTable.toDF.filter($"id" > 99997).show
+------+----+
|    id|rand|
+------+----+
|100544|   5|
|100795|   1|
|101090|   5|
|101499|   1|
|101963|   7|

(snip)

def merge(source: DataFrame, condition: Column): DeltaMergeBuilder = {
  DeltaMergeBuilder(this, source, condition)
}

def update(set: Map[String, Column]): DeltaMergeBuilder = {
  addUpdateClause(set)
}

private def addUpdateClause(set: Map[String, Column]): DeltaMergeBuilder = {
  if (set.isEmpty && matchCondition.isEmpty) {
    // Nothing to update = no need to add an update clause
    mergeBuilder
  } else {
    val setActions = set.toSeq
    val updateActions = MergeIntoClause.toActions(
      colNames = setActions.map(x => UnresolvedAttribute.quotedString(x._1)),
      exprs = setActions.map(x => x._2.expr),
      isEmptySeqEqualToStar = false)
    val updateClause = MergeIntoUpdateClause(matchCondition.map(_.expr), updateActions)
    mergeBuilder.withClause(updateClause)
  }
}

private[delta] def withClause(clause: MergeIntoClause): DeltaMergeBuilder = {
  new DeltaMergeBuilder(
    this.targetTable, this.source, this.onCondition, this.whenClauses :+ clause)
}

def execute(): Unit = {
  val sparkSession = targetTable.toDF.sparkSession
  val resolvedMergeInto =
    MergeInto.resolveReferences(mergePlan)(tryResolveReferences(sparkSession) _)
  if (!resolvedMergeInto.resolved) {
    throw DeltaErrors.analysisException("Failed to resolve\n", plan = Some(resolvedMergeInto))
  }
  // Preprocess the actions and verify
  val mergeIntoCommand = PreprocessTableMerge(sparkSession.sessionState.conf)(resolvedMergeInto)
  sparkSession.sessionState.analyzer.checkAnalysis(mergeIntoCommand)
  mergeIntoCommand.run(sparkSession)
}

val matchedRowCounts = collectTouchedFiles.groupBy(ROW_ID_COL).agg(sum("one").as("count"))
if (matchedRowCounts.filter("count > 1").count() != 0) {
  throw DeltaErrors.multipleSourceRowMatchingTargetRowInMergeException(spark)
}

scala> val deltaTable = DeltaTable.forPath("/tmp/delta-table")
scala> deltaTable.history.show
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+
|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|isolationLevel|isBlindAppend|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+
|     13|2020-02-26 23:19:17|  null|    null|    MERGE|[predicate -> (ol...|null|    null|     null|         12|          null|        false|
|     12|2020-02-26 23:16:13|  null|    null|   DELETE|[predicate -> ["(...|null|    null|     null|         11|          null|        false|
|     11|2020-02-26 23:16:00|  null|    null|   UPDATE|[predicate -> ((i...|null|    null|     null|         10|          null|        false|
|     10|2020-02-26 23:12:48|  null|    null|    WRITE|[mode -> Append, ...|null|    null|     null|          9|          null|         true|
|      9|2020-02-26 23:12:33|  null|    null|    WRITE|[mode -> Overwrit...|null|    null|     null|          8|          null|        false|
|      8|2020-02-26 23:11:48|  null|    null|    WRITE|[mode -> Append, ...|null|    null|     null|          7|          null|         true|

(snip)

scala> spark.read.format("delta").option("versionAsOf", 8).load("/tmp/delta-table").show

val timeTravelByParams = getTimeTravelVersion(parameters)

df1 = spark.read.format("delta").option("timestampAsOf", timestamp_string).load("/delta/events")

override def run(sparkSession: SparkSession): Seq[Row] = {
  deltaLog.withNewTransaction { txn =>
    val actions = write(txn, sparkSession)
    val operation = DeltaOperations.Write(mode, Option(partitionColumns), options.replaceWhere)
    txn.commit(actions, operation)
  }
  Seq.empty
}

  def commit(actions: Seq[Action], op: DeltaOperations.Operation): Long = recordDeltaOperation(
      deltaLog,
      "delta.commit") {
    val version = try {

(snip)

val version = try {
  // Try to commit at the next version.
  var finalActions = prepareCommit(actions, op)

// Find the isolation level to use for this commit
val noDataChanged = actions.collect { case f: FileAction => f.dataChange }.forall(_ == false)
val isolationLevelToUse = if (noDataChanged) {
  // If no data has changed (i.e. its is only being rearranged), then SnapshotIsolation
  // provides Serializable guarantee. Hence, allow reduced conflict detection by using
  // SnapshotIsolation of what the table isolation level is.
  SnapshotIsolation
} else {
  Serializable
}

val isolationLevelToUse = if (noDataChanged) {
  // If no data has changed (i.e. its is only being rearranged), then SnapshotIsolation
  // provides Serializable guarantee. Hence, allow reduced conflict detection by using
  // SnapshotIsolation of what the table isolation level is.
  SnapshotIsolation
} else {
  Serializable
}

val commitVersion = doCommit(snapshot.version + 1, finalActions, 0, isolationLevelToUse)

private def doCommit(
    attemptVersion: Long,
    actions: Seq[Action],
    attemptNumber: Int,
    isolationLevel: IsolationLevel): Long = deltaLog.lockInterruptibly {
  try {
    logDebug(
      s"Attempting to commit version $attemptVersion with ${actions.size} actions with " +
        s"$isolationLevel isolation level")

    deltaLog.store.write(
      deltaFile(deltaLog.logPath, attemptVersion),
      actions.map(_.json).toIterator)

} catch {
  case e: java.nio.file.FileAlreadyExistsException =>
    checkAndRetry(attemptVersion, actions, attemptNumber, isolationLevel)
}

protected def checkAndRetry(
    checkVersion: Long,
    actions: Seq[Action],
    attemptNumber: Int,
    commitIsolationLevel: IsolationLevel): Long = recordDeltaOperation(
      deltaLog,
      "delta.commit.retry",
      tags = Map(TAG_LOG_STORE_CLASS -> deltaLog.store.getClass.getName)) {

  import _spark.implicits._

  val nextAttemptVersion = getNextAttemptVersion(checkVersion)

    (checkVersion until nextAttemptVersion).foreach { version =>
      // Actions of a commit which went in before ours
      val winningCommitActions =
        deltaLog.store.read(deltaFile(deltaLog.logPath, version)).map(Action.fromJson)

(snip)

val metadataUpdates = winningCommitActions.collect { case a: Metadata => a }
val removedFiles = winningCommitActions.collect { case a: RemoveFile => a }
val txns = winningCommitActions.collect { case a: SetTransaction => a }
val protocol = winningCommitActions.collect { case a: Protocol => a }
val commitInfo = winningCommitActions.collectFirst { case a: CommitInfo => a }.map(
  ci => ci.copy(version = Some(version)))

val blindAppendAddedFiles = mutable.ArrayBuffer[AddFile]()
val changedDataAddedFiles = mutable.ArrayBuffer[AddFile]()

val isBlindAppendOption = commitInfo.flatMap(_.isBlindAppend)
if (isBlindAppendOption.getOrElse(false)) {
  blindAppendAddedFiles ++= winningCommitActions.collect { case a: AddFile => a }
} else {
  changedDataAddedFiles ++= winningCommitActions.collect { case a: AddFile => a }
}

if (protocol.nonEmpty) {
  protocol.foreach { p =>
    deltaLog.protocolRead(p)
    deltaLog.protocolWrite(p)
  }
  actions.foreach {
    case Protocol(_, _) => throw new ProtocolChangedException(commitInfo)
    case _ =>
  }
}