start implementing refreshes orchestration

Author: edgao

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcDestinationHandler.kt

@@ -325,6 +325,10 @@ abstract class JdbcDestinationHandler<DestinationState>(
                     streamConfig,
                     finalTableDefinition.isPresent,
                     initialRawTableState,
+                    // TODO fix this
+                    // for now, no JDBC destinations actually do refreshes
+                    // so this is just to make our code compile
+                    InitialRawTableStatus(false, false, Optional.empty()),
                     isSchemaMismatch,
                     isFinalTableEmpty,
                     destinationState

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/kotlin/io/airbyte/integrations/base/destination/operation/AbstractStreamOperation.kt

@@ -18,6 +18,9 @@ interface StorageOperation<Data> {
      * Prepare staging area which cloud be creating any object storage, temp tables or file storage.
      * Similar to [createFinalTable], accepts a [suffix] parameter, which should be used in conjunction
      * with [overwriteStage].
+     *
+     * @param replace If true, then replace existing resources with empty e.g. tables. If false,
+     * then leave existing resources untouched.
      */
     fun prepareStage(streamId: StreamId, suffix: String, replace: Boolean = false)
 
@@ -40,8 +43,10 @@ interface StorageOperation<Data> {
      *
      * [AbstractStreamOperation] is responsible for orchestrating the stages so that the temp stage
      * always contains exactly one generation.
+     *
+     * @return The generation ID of a record in the stage, or `null` if the stage is empty.
      */
-    fun getStageGeneration(streamId: StreamId, suffix: String): Long
+    fun getStageGeneration(streamId: StreamId, suffix: String): Long?
 
     /** Delete previously staged data, using deterministic information from streamId. */
     fun cleanupStage(streamId: StreamId)

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/kotlin/io/airbyte/integrations/base/destination/operation/StorageOperation.kt

@@ -7,7 +7,13 @@ package io.airbyte.integrations.base.destination.typing_deduping
 data class DestinationInitialStatus<DestinationState>(
     val streamConfig: StreamConfig,
     val isFinalTablePresent: Boolean,
+    // TODO we should probably make this nullable, then delete InitialRawTableStatus.rawTableExists
     val initialRawTableStatus: InitialRawTableStatus,
+    /**
+     * The state of the temp raw table, or null if there is no temp raw table
+     * at the start of the sync.
+     */
+    val initialTempRawTableStatus: InitialRawTableStatus,
     val isSchemaMismatch: Boolean,
     val isFinalTableEmpty: Boolean,
     val destinationState: DestinationState,

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/kotlin/io/airbyte/integrations/base/destination/typing_deduping/DestinationInitialStatus.kt

@@ -8,6 +8,19 @@ import java.util.*
 
 data class InitialRawTableStatus(
     val rawTableExists: Boolean,
+    /**
+     * Whether there were any records with null `_airbyte_loaded_at`, at the time
+     * that this status was fetched.
+     */
     val hasUnprocessedRecords: Boolean,
+    // TODO Make maxProcessedTimestamp just `Instant?` instead of Optional
+    /**
+     * The highest timestamp such that all records in
+     * `SELECT * FROM raw_table WHERE _airbyte_extracted_at <= ?`
+     * have a nonnull `_airbyte_loaded_at`.
+     *
+     * Destinations MAY use this value to only run T+D on records with
+     * `_airbyte_extracted_at > ?` (note the strictly-greater comparison).
+     */
     val maxProcessedTimestamp: Optional<Instant>
 )

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/kotlin/io/airbyte/integrations/base/destination/typing_deduping/InitialRawTableStatus.kt

@@ -42,8 +42,9 @@ data class StreamId(
         return "$quote$finalNamespace$quote.$quote$finalName$suffix$quote"
     }
 
-    fun rawTableId(quote: String): String {
-        return "$quote$rawNamespace$quote.$quote$rawName$quote"
+    @JvmOverloads
+    fun rawTableId(quote: String, suffix: String = ""): String {
+        return "$quote$rawNamespace$quote.$quote$rawName$suffix$quote"
     }
 
     fun finalName(quote: String): String {