scala> val df = spark.read.format("delta").load("/tmp/delta/wordcount")
df: org.apache.spark.sql.DataFrame = [value: string, count: bigint]

scala> df.show
+-----+-----+
|value|count|
+-----+-----+
| fuga|    5|
| hoge|    3|
+-----+-----+

scala> val df = spark.read.format("delta").load("/tmp/delta/wordcount_per_line")
df: org.apache.spark.sql.DataFrame = [unixtime: bigint, count: bigint]

scala> df.show
+-------------+-----+
|     unixtime|count|
+-------------+-----+
|1605968491821|    2|
+-------------+-----+

scala> val df = spark.read.format("parquet").load("/opt/spark/default/examples/src/main/resources/users.parquet")
scala> df.show
+------+--------------+----------------+
|  name|favorite_color|favorite_numbers|
+------+--------------+----------------+
|Alyssa|          null|  [3, 9, 15, 20]|
|   Ben|           red|              []|
+------+--------------+----------------+

scala> df.write.format("delta").save("/tmp/delta/users")

scala> df.printSchema
root
 |-- name: string (nullable = true)
 |-- favorite_color: string (nullable = true)
 |-- favorite_numbers: array (nullable = true)
 |    |-- element: integer (containsNull = true)

scala> case class User(name: String, favorite_color: String, favorite_numbers: Array[Int])

scala> val addRecord = Seq(User("Bob", "yellow", Array(1,2))).toDF

scala> addRecord.write.format("delta").mode("append").save("/tmp/delta/users")

val maxFilesPerTrigger = options.get(MAX_FILES_PER_TRIGGER_OPTION).map { str =>
  Try(str.toInt).toOption.filter(_ > 0).getOrElse {
    throw DeltaErrors.illegalDeltaOptionException(
      MAX_FILES_PER_TRIGGER_OPTION, str, "must be a positive integer")
  }
}

  def toReadLimit: ReadLimit = {
    if (options.maxFilesPerTrigger.isDefined && options.maxBytesPerTrigger.isDefined) {
      CompositeLimit(
        ReadMaxBytes(options.maxBytesPerTrigger.get),
        ReadLimit.maxFiles(options.maxFilesPerTrigger.get).asInstanceOf[ReadMaxFiles])
    } else if (options.maxBytesPerTrigger.isDefined) {
      ReadMaxBytes(options.maxBytesPerTrigger.get)
    } else {
      ReadLimit.maxFiles(
        options.maxFilesPerTrigger.getOrElse(DeltaOptions.MAX_FILES_PER_TRIGGER_OPTION_DEFAULT))
    }
  }
}

scala> val df = spark.read.format("parquet").load("/opt/spark/default/examples/src/main/resources/users.parquet")
scala> df.show
+------+--------------+----------------+
|  name|favorite_color|favorite_numbers|
+------+--------------+----------------+
|Alyssa|          null|  [3, 9, 15, 20]|
|   Ben|           red|              []|
+------+--------------+----------------+

scala> df.write.format("delta").partitionBy("name").save("/tmp/delta/partitioned_users")

scala> val updateRecord = Seq(User("Ben", "green", Array(1,2))).toDF

scala> updateRecord
         .write
         .format("delta")
         .mode("overwrite")
         .option("replaceWhere", "name == 'Ben'")
         .save("/tmp/delta/partitioned_users")

scala> val df = spark.read.format("parquet").load("/opt/spark/default/examples/src/main/resources/users.parquet")
scala> df.show
+------+--------------+----------------+
|  name|favorite_color|favorite_numbers|
+------+--------------+----------------+
|Alyssa|          null|  [3, 9, 15, 20]|
|   Ben|           red|              []|
+------+--------------+----------------+

scala> df.write.format("delta").save("/tmp/delta/users_for_delete")

scala> import io.delta.tables._
scala> val deltaTable = DeltaTable.forPath(spark, "/tmp/delta/users_for_delete")

scala> deltaTable.delete("name == 'Ben'")

scala> import io.delta.tables._
scala> val deltaTable = DeltaTable.forPath(spark, "/tmp/delta/users_for_delete2")

scala> deltaTable.delete("name == 'Ben'")

scala> deltaTable.toDF.show
+------+--------------+----------------+
|  name|favorite_color|favorite_numbers|
+------+--------------+----------------+
|Alyssa|          null|  [3, 9, 15, 20]|
+------+--------------+----------------+

/** A check on the source table that disallows deletes on the source data. */
private val ignoreChanges = options.ignoreChanges || ignoreFileDeletion
--> レコードの変更（ファイル変更）を無視するかどうか

/** A check on the source table that disallows commits that only include deletes to the data. */
private val ignoreDeletes = options.ignoreDeletes || ignoreFileDeletion || ignoreChanges
--> レコードの削除（ファイル削除）を無視するかどうか

private val excludeRegex: Option[Regex] = options.excludeRegex
--> ADDファイルのリストする際、無視するファイルを指定する

override val schema: StructType = deltaLog.snapshot.metadata.schema
--> 対象テーブルのスキーマ

(snip)

private val tableId = deltaLog.snapshot.metadata.id
--> テーブルのID

private var previousOffset: DeltaSourceOffset = null
--> バッチの取得時のオフセットを保持する

// A metadata snapshot when starting the query.
private var initialState: DeltaSourceSnapshot = null
private var initialStateVersion: Long = -1L
--> org.apache.spark.sql.delta.sources.DeltaSource#getBatch などから呼ばれる、スナップショットを保持するための変数

val model = if (param.modelSnapshot.isDefined) {
  Module.load[Float](param.modelSnapshot.get)
} else {
  if (param.graphModel) {
    LeNet5.graph(classNum = 10)
  } else {
    Engine.getEngineType() match {
      case MklBlas => LeNet5(10)
      case MklDnn => LeNet5.dnnGraph(param.batchSize / Engine.nodeNumber(), 10)
    }
  }
}

val optimizer = Optimizer(
  model = model,
  dataset = trainSet,
  criterion = criterion)

dataset match {
  case d: DistributedDataSet[_] =>
    Engine.getOptimizerVersion() match {
      case OptimizerV1 =>
        new DistriOptimizer[T](
          _model = model,
          _dataset = d.toDistributed().asInstanceOf[DistributedDataSet[MiniBatch[T]]],
          _criterion = criterion
        ).asInstanceOf[Optimizer[T, D]]
      case OptimizerV2 =>
        new DistriOptimizerV2[T](
          _model = model,
          _dataset = d.toDistributed().asInstanceOf[DistributedDataSet[MiniBatch[T]]],
          _criterion = criterion
        ).asInstanceOf[Optimizer[T, D]]
    }
  case d: LocalDataSet[_] =>
    new LocalOptimizer[T](
      model = model,
      dataset = d.toLocal().asInstanceOf[LocalDataSet[MiniBatch[T]]],
      criterion = criterion
    ).asInstanceOf[Optimizer[T, D]]
  case _ =>
    throw new UnsupportedOperationException
}

optimizer
  .setValidation(
    trigger = Trigger.everyEpoch,
    dataset = validationSet,
    vMethods = Array(new Top1Accuracy, new Top5Accuracy[Float], new Loss[Float]))
  .setOptimMethod(optimMethod)
  .setEndWhen(Trigger.maxEpoch(param.maxEpoch))
  .optimize()

prepareInput()

if (!dataset.toDistributed().isCached) {
  DistriOptimizer.logger.info("caching training rdd ...")
  DistriOptimizer.prepareInput(this.dataset, this.validationDataSet)
}

private[bigdl] def prepareInput[T: ClassTag](dataset: DataSet[MiniBatch[T]],
  validationDataSet: Option[DataSet[MiniBatch[T]]]): Unit = {
  dataset.asInstanceOf[DistributedDataSet[MiniBatch[T]]].cache()
  if (validationDataSet.isDefined) {
    validationDataSet.get.toDistributed().cache()
  }
}

def cache(): Unit = {
  if (originRDD() != null) {
    originRDD().count()
  }
  isCached = true
}

val trainSet = DataSet.array(load(trainData, trainLabel), sc) ->
  BytesToGreyImg(28, 28) -> GreyImgNormalizer(trainMean, trainStd) -> GreyImgToBatch(
  param.batchSize)

def array[T: ClassTag](localData: Array[T], sc: SparkContext): DistributedDataSet[T] = {
  val nodeNumber = Engine.nodeNumber()
  new CachedDistriDataSet[T](
    sc.parallelize(localData, nodeNumber)
      // Keep this line, or the array will be send to worker every time
      .coalesce(nodeNumber, true)
      .mapPartitions(iter => {
        Iterator.single(iter.toArray)
      }).setName("cached dataset")
      .cache()
  )
}

_dataset = (DataSet.rdd(sampleRDD) ->

def rdd[T: ClassTag](data: RDD[T], partitionNum: Int = Engine.nodeNumber()
  ): DistributedDataSet[T] = {
  new CachedDistriDataSet[T](
    data.coalesce(partitionNum, true)
      .mapPartitions(iter => {
        Iterator.single(iter.toArray)
      }).setName("cached dataset")
      .cache()
  )
}

scala> spark.sparkContext.hadoopConfiguration.set("fs.s3a.aws.credentials.provider", "com.amazonaws.auth.DefaultAWSCredentialsProviderChain")

class DeltaSparkSessionExtension extends (SparkSessionExtensions => Unit) {
  override def apply(extensions: SparkSessionExtensions): Unit = {
    extensions.injectParser { (session, parser) =>
      new DeltaSqlParser(parser)
    }

(snip)

class DeltaSqlParser(val delegate: ParserInterface) extends ParserInterface {
  private val builder = new DeltaSqlAstBuilder

(snip)

def sql(sqlText: String): DataFrame = withActive {
  val tracker = new QueryPlanningTracker
  val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
    sessionState.sqlParser.parsePlan(sqlText)
  }
  Dataset.ofRows(self, plan, tracker)
}

override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
  builder.visit(parser.singleStatement()) match {
    case plan: LogicalPlan => plan
    case _ => delegate.parsePlan(sqlText)
  }
}

override def createTable(
    ident: Identifier,
    schema: StructType,
    partitions: Array[Transform],
    properties: util.Map[String, String]): Table = {
  if (DeltaSourceUtils.isDeltaDataSourceName(getProvider(properties))) {
    createDeltaTable(
      ident, schema, partitions, properties, sourceQuery = None, TableCreationModes.Create)
  } else {
    super.createTable(ident, schema, partitions, properties)
  }
}

df.write.format("delta").saveAsTable("events")      // create table in the metastore

df.write.format("delta").save("/delta/events")  // create table by path

spark.sql(s"""MERGE INTO $tableName USING newData
    ON ${tableName}.id = newData.id
    WHEN MATCHED THEN
      UPDATE SET ${tableName}.id = newData.id
    WHEN NOT MATCHED THEN INSERT *
""")

scala> import io.delta.tables.DeltaTable
scala> val tbl = DeltaTable.forName(tableName)

spark.sql(s"CREATE TABLE $tableName(id LONG) USING delta TBLPROPERTIES ('delta.logRetentionDuration' = 'interval 1 day', 'delta.deletedFileRetentionDuration' = 'interval 1 day')")
spark.sql(s"ALTER TABLE $tableName SET TBLPROPERTIES ('delta.logRetentionDuration' = 'interval 1 day', 'delta.deletedFileRetentionDuration' = 'interval 1 day')")

df.write.option("spark.databricks.delta.commitInfo.userMetadata", "test").format("delta").mode("append").save("/tmp/test")

scala> spark.sql("SET spark.databricks.delta.commitInfo.userMetadata=test")
scala> spark.sql(s"INSERT INTO $tableName VALUES 0, 1, 2, 3, 4")

scala> val home = sys.env("HOME")
scala> val imageDir = "Downloads/mnist_jpg/trainingSet/trainingSet/0"
scala> val df = spark.read.format("image").option("dropInvalid", true).load(home + "/" + imageDir)

scala> df.select("image.origin", "image.width", "image.height").show(3, truncate=false)
+-------------------------------------------------------------------------------+-----+------+
|origin                                                                         |width|height|
+-------------------------------------------------------------------------------+-----+------+
|file:///home/centos/Downloads/mnist_jpg/trainingSet/trainingSet/0/img_20188.jpg|28   |28    |
|file:///home/centos/Downloads/mnist_jpg/trainingSet/trainingSet/0/img_12634.jpg|28   |28    |
|file:///home/centos/Downloads/mnist_jpg/trainingSet/trainingSet/0/img_26812.jpg|28   |28    |
+-------------------------------------------------------------------------------+-----+------+

scala> df.printSchema
root
 |-- image: struct (nullable = true)
 |    |-- origin: string (nullable = true)
 |    |-- height: integer (nullable = true)
 |    |-- width: integer (nullable = true)
 |    |-- nChannels: integer (nullable = true)
 |    |-- mode: integer (nullable = true)
 |    |-- data: binary (nullable = true)

scala> val deltaImagePath = "/tmp/delta-lake-image"
scala> df.write.format("delta").save(deltaImagePath)

scala> import io.delta.tables._
scala> import org.apache.spark.sql.functions._
scala> val deltaTable = DeltaTable.forPath(deltaImagePath)

scala> val targetDf = df.select($"image.origin", $"image")
scala> val targetImagePath = "/tmp/delta-table"
scala> targetDf.write.format("delta").save(targetImagePath)
scala> val sourcePath = "Downloads/mnist_jpg/trainingSet/trainingSet/1"
scala> val sourceDf = spark.read.format("image").option("dropInvalid", true).load(home + "/" + sourcePath).select($"image.origin", $"image")

scala> val deltaTable = DeltaTable.forPath(targetImagePath)
scala> deltaTable
        .as("target")
        .merge(
          sourceDf.as("source"),
          "target.origin = source.origin")
        .whenMatched
        .updateExpr(Map(
          "image" -> "source.image"))
        .whenNotMatched
        .insertExpr(Map(
          "origin" -> "source.origin",
          "image" -> "source.image"))
        .execute()

scala> deltaTable.toDF.where($"origin" contains "trainingSet/1").select("image.origin", "image.width", "image.height").show(3, truncate=false)
+-------------------------------------------------------------------------------+-----+------+
|origin                                                                         |width|height|
+-------------------------------------------------------------------------------+-----+------+
|file:///home/centos/Downloads/mnist_jpg/trainingSet/trainingSet/1/img_10266.jpg|28   |28    |
|file:///home/centos/Downloads/mnist_jpg/trainingSet/trainingSet/1/img_10665.jpg|28   |28    |
|file:///home/centos/Downloads/mnist_jpg/trainingSet/trainingSet/1/img_10772.jpg|28   |28    |
+-------------------------------------------------------------------------------+-----+------+
only showing top 3 rows

scala> import org.apache.spark.sql.SaveMode._
scala> import org.apache.hudi.DataSourceReadOptions._
scala> import org.apache.hudi.DataSourceWriteOptions._
scala> import org.apache.hudi.config.HoodieWriteConfig._

scala> val tableName = "hudi_images"
scala> val basePath = "file:///tmp/hudi_images"
scala> val home = sys.env("HOME")
scala> val imageDir = "Downloads/mnist_jpg/trainingSet/trainingSet/*"
scala> val df = spark.read.format("image").option("dropInvalid", true).load(home + "/" + imageDir)

scala> import org.apache.spark.sql.functions.{col, udf}
scala> val partitionPath = udf((str: String) => {
         str.split("/").takeRight(2).head
       })
scala> val hudiDf = df.select($"image.origin", partitionPath($"image.origin") as "partitionpath", $"*")

scala> hudiDf.write.format("hudi").
         option(TABLE_NAME, tableName).
         option(PRECOMBINE_FIELD_OPT_KEY, "origin").
         option(RECORDKEY_FIELD_OPT_KEY, "origin").
         option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
         mode(Overwrite).
         save(basePath)

scala> val hudiImageDf = spark.
         read.
         format("hudi").
         load(basePath + "/*")

scala> hudiImageDf.printSchema
root
 |-- _hoodie_commit_time: string (nullable = true)
 |-- _hoodie_commit_seqno: string (nullable = true)
 |-- _hoodie_record_key: string (nullable = true)
 |-- _hoodie_partition_path: string (nullable = true)
 |-- _hoodie_file_name: string (nullable = true)
 |-- origin: string (nullable = true)
 |-- partitionpath: string (nullable = true)
 |-- image: struct (nullable = true)
 |    |-- origin: string (nullable = true)
 |    |-- height: integer (nullable = true)
 |    |-- width: integer (nullable = true)
 |    |-- nChannels: integer (nullable = true)
 |    |-- mode: integer (nullable = true)
 |    |-- data: binary (nullable = true)

scala> val updateImageDir = "Downloads/mnist_jpg/trainingSet/trainingSet/0"
scala> val updateDf = spark.read.format("image").option("dropInvalid", true).load(home + "/" + updateImageDir)
scala> val updateHudiDf = updateDf.select($"image.origin", partitionPath($"image.origin") as "partitionpath", $"*")
scala> updateHudiDf.write.format("hudi").
         option(TABLE_NAME, tableName).
         option(PRECOMBINE_FIELD_OPT_KEY, "origin").
         option(RECORDKEY_FIELD_OPT_KEY, "origin").
         option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
         mode(Append).
         save(basePath)

scala> val updatedHudiImageDf = spark.
         read.
         format("hudi").
         load(basePath + "/*")

scala> updatedHudiImageDf.select($"_hoodie_commit_time", $"_hoodie_commit_seqno", $"_hoodie_partition_path").filter("partitionpath == 1").show(3)
+-------------------+--------------------+----------------------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_partition_path|
+-------------------+--------------------+----------------------+
|     20200517135104|  20200517135104_1_1|                     1|
|     20200517135104| 20200517135104_1_12|                     1|
|     20200517135104| 20200517135104_1_45|                     1|
+-------------------+--------------------+----------------------+
only showing top 3 rows


scala> updatedHudiImageDf.select($"_hoodie_commit_time", $"_hoodie_commit_seqno", $"_hoodie_partition_path").filter("partitionpath == 0").show(3)
+-------------------+--------------------+----------------------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_partition_path|
+-------------------+--------------------+----------------------+
|     20200517140006|20200517140006_0_...|                     0|
|     20200517140006|20200517140006_0_...|                     0|
|     20200517140006|20200517140006_0_...|                     0|
+-------------------+--------------------+----------------------+
only showing top 3 rows