TransactionalWrite¶

TransactionalWrite is an abstraction of optimistic transactional writers that can write a structured query out to a Delta table.

Contract¶

DeltaLog¶

deltaLog: DeltaLog

DeltaLog (of a delta table) that this transaction is changing

Used when:

• OptimisticTransactionImpl is requested to prepare a commit, doCommit, checkAndRetry, and perform post-commit operations (and execute delta log checkpoint)
• ConvertToDeltaCommand is executed
• DeltaCommand is requested to buildBaseRelation and commitLarge
• MergeIntoCommand is executed
• TransactionalWrite is requested to write a structured query out to a delta table
• GenerateSymlinkManifest post-commit hook is executed
• ImplicitMetadataOperation is requested to updateMetadata
• DeltaSink is requested to addBatch

Metadata¶

metadata: Metadata

Metadata (of the delta table) that this transaction is changing

Protocol¶

protocol: Protocol

Protocol (of the delta table) that this transaction is changing

Used when:

• OptimisticTransactionImpl is requested to updateMetadata, verifyNewMetadata and prepareCommit
• ConvertToDeltaCommand is executed

Snapshot¶

snapshot: Snapshot

Snapshot (of the delta table) that this transaction is reading at

Implementations¶

• OptimisticTransaction

spark.databricks.delta.history.metricsEnabled¶

With spark.databricks.delta.history.metricsEnabled configuration property enabled, TransactionalWrite creates a BasicWriteJobStatsTracker (Spark SQL) and registers SQL metrics (when requested to write data out).

hasWritten Flag¶

hasWritten: Boolean = false

TransactionalWrite uses hasWritten internal registry to prevent OptimisticTransactionImpl from updating metadata after having written out files.

hasWritten is initially false and changes to true after having data written out.

Writing Data Out¶

writeFiles(
  data: Dataset[_]): Seq[FileAction]  // (1)!
writeFiles(
  data: Dataset[_],
  writeOptions: Option[DeltaOptions]): Seq[FileAction]
writeFiles(
  data: Dataset[_],
  writeOptions: Option[DeltaOptions],
  additionalConstraints: Seq[Constraint]): Seq[FileAction]  // (2)!
writeFiles(
  data: Dataset[_],
  additionalConstraints: Seq[Constraint]): Seq[FileAction]

1. Uses no additionalConstraints
2. writeOptions are ignored

writeFiles writes the given data to a delta table. writeFiles returns AddFiles (that are the new files added to a reservoir).

writeFiles creates a DeltaInvariantCheckerExec and a DelayedCommitProtocol to write out files to the data path (of the DeltaLog).

writeFiles is executed within SQLExecution.withNewExecutionId.

In the end, writeFiles returns the addedStatuses of the DelayedCommitProtocol committer.

Internally, writeFiles turns the hasWritten flag on (true).

writeFiles normalize the given data dataset (based on the partitionColumns of the Metadata).

writeFiles getPartitioningColumns based on the partitionSchema of the Metadata.

DelayedCommitProtocol Committer¶

writeFiles creates a DelayedCommitProtocol committer for the data path (of the DeltaLog).

Constraints¶

writeFiles collects constraintss from the table metadata and the generated columns.

DeltaInvariantCheckerExec¶

writeFiles requests a new Execution ID (that is used to track all Spark jobs of FileFormatWriter.write in Spark SQL) with a physical query plan of a new DeltaInvariantCheckerExec unary physical operator (with the executed plan of the normalized query execution as the child operator)

Creating Committer¶

getCommitter(
  outputPath: Path): DelayedCommitProtocol

getCommitter creates a new DelayedCommitProtocol with the delta job ID and the given outputPath (and no random prefix).

getPartitioningColumns¶

getPartitioningColumns(
  partitionSchema: StructType,
  output: Seq[Attribute],
  colsDropped: Boolean): Seq[Attribute]

getPartitioningColumns...FIXME

normalizeData¶

normalizeData(
  data: Dataset[_],
  partitionCols: Seq[String]): (QueryExecution, Seq[Attribute])

normalizeData...FIXME

makeOutputNullable¶

makeOutputNullable(
  output: Seq[Attribute]): Seq[Attribute]

makeOutputNullable...FIXME

Usage¶

writeFiles is used when:

• DeleteCommand, MergeIntoCommand, OptimizeTableCommand, UpdateCommand and WriteIntoDelta commands are executed
• DeltaSink is requested to add a streaming micro-batch