WriteIntoDelta Command¶

WriteIntoDelta is a Delta command that can write data(frame) transactionally into a delta table.

WriteIntoDelta is a RunnableCommand (Spark SQL) logical operator.

Creating Instance¶

WriteIntoDelta takes the following to be created:

• DeltaLog
• SaveMode
• DeltaOptions
• Names of the partition columns
• Configuration
• Data (DataFrame)

WriteIntoDelta is created when:

• DeltaLog is requested to create an insertable HadoopFsRelation (when DeltaDataSource is requested to create a relation as a CreatableRelationProvider or a RelationProvider)
• DeltaCatalog is requested to createDeltaTable
• WriteIntoDeltaBuilder is requested to buildForV1Write
• CreateDeltaTableCommand command is executed
• DeltaDataSource is requested to create a relation (for writing) (as a CreatableRelationProvider)

Configuration¶

WriteIntoDelta is given a configuration when created as follows:

• Always empty for DeltaLog
• Always empty for DeltaDataSource
• Existing properties of a delta table in DeltaCatalog (with the comment key based on the value in the catalog)
• Existing configuration (of the Metadata of the Snapshot of the DeltaLog) for WriteIntoDeltaBuilder
• Existing properties of a delta table for CreateDeltaTableCommand (with the comment key based on the value in the catalog)

ImplicitMetadataOperation¶

WriteIntoDelta is an operation that can update metadata (schema and partitioning) of the delta table.

Executing Command¶

run(
  sparkSession: SparkSession): Seq[Row]

run is part of the RunnableCommand (Spark SQL) abstraction.

run requests the DeltaLog to start a new transaction.

run writes and requests the OptimisticTransaction to commit (with DeltaOperations.Write operation with the SaveMode, partition columns, replaceWhere and userMetadata).

write¶

write(
  txn: OptimisticTransaction,
  sparkSession: SparkSession): Seq[Action]

write checks out whether the write operation is to a delta table that already exists. If so (i.e. the readVersion of the transaction is above -1), write branches per the SaveMode:

• For ErrorIfExists, write throws an AnalysisException.

  [path] already exists.
  

• For Ignore, write does nothing and returns back with no Actions.

• For Overwrite, write requests the DeltaLog to assert being removable

write updateMetadata (with rearrangeOnly option).

write...FIXME

write is used when:

• CreateDeltaTableCommand is executed
• WriteIntoDelta is executed

Demo¶

import org.apache.spark.sql.delta.commands.WriteIntoDelta
import org.apache.spark.sql.delta.DeltaLog
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.delta.DeltaOptions
val tableName = "/tmp/delta/t1"
val data = spark.range(5).toDF
val writeCmd = WriteIntoDelta(
  deltaLog = DeltaLog.forTable(spark, tableName),
  mode = SaveMode.Overwrite,
  options = new DeltaOptions(Map.empty[String, String], spark.sessionState.conf),
  partitionColumns = Seq.empty[String],
  configuration = Map.empty[String, String],
  data)

// Review web UI @ http://localhost:4040

writeCmd.run(spark)

canOverwriteSchema¶

canOverwriteSchema: Boolean

canOverwriteSchema is part of the ImplicitMetadataOperation abstraction.

canOverwriteSchema is true when all the following hold:

1. canOverwriteSchema is enabled (true) (in the DeltaOptions)
2. This WriteIntoDelta is overwrite operation
3. replaceWhere option is not defined (in the DeltaOptions)

isOverwriteOperation¶

isOverwriteOperation: Boolean

isOverwriteOperation is true for the SaveMode to be SaveMode.Overwrite.

isOverwriteOperation is used when:

• WriteIntoDelta is requested for the canOverwriteSchema and to write

extractConstraints¶

extractConstraints(
  sparkSession: SparkSession,
  expr: Seq[Expression]): Seq[Constraint]

extractConstraints...FIXME

extractConstraints is used when:

• WriteIntoDelta is requested to write data out (with SaveMode.Overwrite mode with replaceWhere option)