[SDP] SystemMetadata

Author: jaceklaskowski

docs/declarative-pipelines/FlowSystemMetadata.md

@@ -1,5 +1,20 @@
 # FlowSystemMetadata
 
+`FlowSystemMetadata` is a [SystemMetadata](SystemMetadata.md) associated with a [Flow](#flow).
+
+## Creating Instance
+
+`FlowSystemMetadata` takes the following to be created:
+
+* <span id="context"> [PipelineUpdateContext](PipelineUpdateContext.md)
+* <span id="flow"> [Flow](Flow.md)
+* <span id="graph"> [DataflowGraph](DataflowGraph.md)
+
+`FlowSystemMetadata` is created when:
+
+* `FlowPlanner` is requested to [plan a StreamingFlow for execution](FlowPlanner.md#plan)
+* `State` is requested to [clear out the state of a flow](State.md#reset)
+
 ## latestCheckpointLocation { #latestCheckpointLocation }
 
 ```scala

docs/declarative-pipelines/PipelineExecution.md

@@ -34,6 +34,8 @@ startPipeline(): Unit
 
 `startPipeline` [resolves and validates the dataflow graph](#resolveGraph) (of this pipeline update).
 
+For a full-refresh update, `startPipeline` [resets the state of all the flows](State.md#reset) in the [DataflowGraph](DataflowGraph.md).
+
 `startPipeline` [materializes the datasets](DatasetManager.md#materializeDatasets) (of this dataflow graph).
 
 `startPipeline` creates a new [TriggeredGraphExecution](#graphExecution) for the materialized dataflow graph.

docs/declarative-pipelines/PipelineUpdateContext.md

@@ -4,6 +4,25 @@
 
 ## Contract (Subset)
 
+### fullRefreshTables { #fullRefreshTables }
+
+```scala
+fullRefreshTables: TableFilter
+```
+
+`TableFilter` of the tables to be fully refreshed in a pipeline update run
+
+See:
+
+* [PipelineUpdateContextImpl](PipelineUpdateContextImpl.md#fullRefreshTables)
+
+Used when:
+
+* `DatasetManager` is requested to [constructFullRefreshSet](DatasetManager.md#constructFullRefreshSet)
+* `PipelineExecution` is requested to [start a pipeline update](PipelineExecution.md#startPipeline)
+* `PipelineUpdateContext` is requested to [refreshFlows](PipelineUpdateContext.md#refreshFlows)
+* `State` is requested to [find the inputs to reset (state of)](State.md#findElementsToReset)
+
 ### refreshTables Table Filter { #refreshTables }
 
 ```scala

docs/declarative-pipelines/State.md

@@ -1,3 +1,48 @@
 # State
 
-`State` is...FIXME
+## Reset State of All Flows { #reset }
+
+```scala
+reset(
+  resolvedGraph: DataflowGraph,
+  env: PipelineUpdateContext): Seq[Input]
+reset(
+  flow: ResolvedFlow,
+  env: PipelineUpdateContext,
+  graph: DataflowGraph): Unit // (1)!
+```
+
+1. A private method
+
+`reset` [finds ResolvedFlows to reset](#findElementsToReset) in the given [DataflowGraph](DataflowGraph.md) (and the [PipelineUpdateContext](PipelineUpdateContext.md)).
+
+!!! info
+    `reset` handles [ResolvedFlow](ResolvedFlow.md)s only.
+
+`reset` prints out the following INFO message to the logs:
+
+```text
+Clearing out state for flow [displayName]
+```
+
+`reset` creates a [FlowSystemMetadata](FlowSystemMetadata.md).
+
+For no [checkpoint directory](FlowSystemMetadata.md#latestCheckpointLocationOpt) available for the [ResolvedFlow](ResolvedFlow.md), `reset` prints out the following INFO message to the logs and exits.
+
+```text
+Skipping resetting flow [identifier]
+since its destination not been previously materialized
+and we can't find the checkpoint location.
+```
+
+Otherwise, when there is a [checkpoint directory](FlowSystemMetadata.md#latestCheckpointLocationOpt) available, `reset` creates a new checkpoint directory (by incrementing the checkpoint number) and prints out the following INFO message to the logs:
+
+```text
+Created new checkpoint for stream [displayName] at [checkpoint_path].
+```
+
+---
+
+`reset` is used when:
+
+* `PipelineExecution` is requested to [run a pipeline update](PipelineExecution.md#startPipeline) (with full-refresh update)

docs/declarative-pipelines/SystemMetadata.md

@@ -0,0 +1,17 @@
+# SystemMetadata
+
+## getLatestCheckpointDir { #getLatestCheckpointDir }
+
+```scala
+getLatestCheckpointDir(
+  rootDir: Path,
+  createNewCheckpointDir: Boolean = false): String
+```
+
+`getLatestCheckpointDir`...FIXME
+
+---
+
+`getLatestCheckpointDir` is used when:
+
+* `FlowSystemMetadata` is requested to [get the latest checkpoint location](FlowSystemMetadata.md#latestCheckpointLocation) and [get the latest checkpoint location if available](#latestCheckpointLocationOpt)