@Singleton
class MySqlGenerator {
    fun createNamespace(namespace: String): String
    fun createTable(stream, tableName, columnMapping, replace): String
    fun dropTable(tableName: TableName): String
    fun copyTable(columnMapping, source, target): String
    fun upsertTable(stream, columnMapping, source, target): String
    fun overwriteTable(source, target): String
    fun alterTable(tableName, added, dropped, modified): Set<String>
    fun countTable(tableName): String
}

fun createTable(stream: DestinationStream, tableName: TableName, ...): String {
    val columnDeclarations = stream.schema.asColumns()
        .map { (name, type) ->
            "${name.quote()} ${columnUtils.toDialectType(type)}"
        }
        .joinToString(",\n")

    return """
        CREATE TABLE ${fullyQualifiedName(tableName)} (
          ${COLUMN_NAME_AB_RAW_ID} VARCHAR NOT NULL,
          ${COLUMN_NAME_AB_EXTRACTED_AT} TIMESTAMP NOT NULL,
          ${COLUMN_NAME_AB_META} JSON NOT NULL,
          ${COLUMN_NAME_AB_GENERATION_ID} INTEGER,
          ${columnDeclarations}
        )
    """.trimIndent().andLog()
}

@Singleton
class MyAirbyteClient(
    private val dataSource: DataSource,
    private val sqlGenerator: MySqlGenerator,
) : TableOperationsClient, TableSchemaEvolutionClient {

    // Namespace operations
    suspend fun createNamespace(namespace: String)
    suspend fun namespaceExists(namespace: String): Boolean

    // Table operations
    suspend fun createTable(stream, tableName, columnMapping, replace)
    suspend fun tableExists(table: TableName): Boolean
    suspend fun dropTable(tableName: TableName)
    suspend fun countTable(tableName: TableName): Long?  // null if not exists

    // Finalization operations
    suspend fun overwriteTable(source, target)  // SWAP/RENAME for truncate
    suspend fun copyTable(columnMapping, source, target)  // Copy data
    suspend fun upsertTable(stream, columnMapping, source, target)  // MERGE for dedupe

    // Metadata
    suspend fun getGenerationId(tableName: TableName): Long
}

// Schema evolution (4 steps)
suspend fun discoverSchema(tableName): TableSchema
fun computeSchema(stream, columnMapping): TableSchema
suspend fun ensureSchemaMatches(stream, tableName, columnMapping)
suspend fun applyChangeset(stream, columnMapping, tableName, expectedColumns, changeset)

override suspend fun createTable(...) {
    execute(sqlGenerator.createTable(...))
}

override suspend fun upsertTable(...) {
    execute(sqlGenerator.upsertTable(...))
}

private suspend fun execute(sql: String) {
    dataSource.connection.use { conn ->
        conn.createStatement().use { stmt ->
            stmt.executeQuery(sql)
        }
    }
}

class MyInsertBuffer(
    private val tableName: TableName,
    private val client: MyAirbyteClient,
    private val flushLimit: Int = 1000,
) {
    private val buffer = mutableListOf<Map<String, AirbyteValue>>()
    private var recordCount = 0

    fun accumulate(recordFields: Map<String, AirbyteValue>) {
        buffer.add(format(recordFields))
        recordCount++

        if (recordCount >= flushLimit) {
            runBlocking { flush() }
        }
    }

    suspend fun flush() {
        if (buffer.isEmpty()) return

        try {
            // Write batch to database
            writeBatchToDatabase(tableName, buffer)
        } finally {
            buffer.clear()
            recordCount = 0
        }
    }

    private fun format(fields: Map<String, AirbyteValue>): Map<String, Any> {
        // Convert AirbyteValue to database types
    }
}

class MyColumnUtils {
    fun toDialectType(type: AirbyteType): String
    fun columnsAndTypes(columns, columnMapping): List<ColumnAndType>
    fun formatColumn(name, type): String
}

fun AirbyteType.toDialectType(): String = when (this) {
    BooleanType -> "BOOLEAN"
    IntegerType -> "BIGINT"
    NumberType -> "DECIMAL(38, 9)"
    StringType -> "VARCHAR"
    DateType -> "DATE"
    TimestampTypeWithTimezone -> "TIMESTAMP WITH TIME ZONE"
    TimestampTypeWithoutTimezone -> "TIMESTAMP"
    is ArrayType -> "JSONB"
    is ObjectType -> "JSONB"
    is UnionType -> "JSONB"
    else -> "VARCHAR"  // Fallback
}

// Snowflake: Add NOT NULL suffix
val typeDecl = if (columnType.nullable) {
    columnType.type  // "VARCHAR"
} else {
    "${columnType.type} NOT NULL"  // "VARCHAR NOT NULL"
}

// ClickHouse: Wrap in Nullable()
val typeDecl = if (columnType.nullable) {
    "Nullable(${columnType.type})"  // "Nullable(String)"
} else {
    columnType.type  // "String"
}

when (stream.minimumGenerationId) {
    0L -> when (stream.importType) {
        Dedupe -> DirectLoadTableDedupStreamLoader       // Temp → MERGE
        else   -> DirectLoadTableAppendStreamLoader      // Direct write
    }
    stream.generationId -> when (stream.importType) {
        Dedupe -> DirectLoadTableDedupTruncateStreamLoader  // Temp → dedupe → SWAP
        else   -> DirectLoadTableAppendTruncateStreamLoader // Temp → SWAP
    }
}

1. CONFIGURATION
   User Config → ConfigFactory → Configuration
   BeanFactory creates all singletons

2. SETUP (Writer.setup())
   - Create all namespaces
   - Gather initial table state

3. STREAM INIT (per stream)
   Writer.createStreamLoader() → select appropriate StreamLoader

   StreamLoader.start():
     - tableExists(finalTable)
     - If exists: ensureSchemaMatches() [schema evolution]
     - If not: createTable(finalTable)
     - If dedupe/truncate: createTable(tempTable)
     - Store target table in streamStateStore

   AggregateFactory.create():
     - Read tableName from streamStateStore
     - Create InsertBuffer(tableName, client)
     - Wrap in Aggregate

4. DATA PROCESSING (automatic)
   Pipeline → Aggregate.accept(record):
     → InsertBuffer.accumulate(record)
     → [auto-flush at threshold]

   Aggregate.flush():
     → InsertBuffer.flush() → write batch to database

5. FINALIZATION (StreamLoader.close())
   If streamCompleted:
     - Dedupe: upsertTable(temp → final) [MERGE]
     - Truncate: overwriteTable(temp → final) [SWAP]
     - Append: nothing (already in final)
   Always:
     - dropTable(tempTable)

Writer
  ├─ depends on: client, statusGatherer, names, streamStateStore
  └─ creates: StreamLoaders

StreamLoader (CDK-provided)
  ├─ depends on: client (TableOperationsClient + TableSchemaEvolutionClient)
  └─ calls: createTable(), ensureSchemaMatches(), upsertTable(), dropTable()

AggregateFactory
  ├─ depends on: client, streamStateStore
  └─ creates: Aggregate + InsertBuffer

InsertBuffer
  ├─ depends on: client, columnUtils
  └─ calls: Only insert operations (NOT upsert/merge/swap)

Client
  ├─ depends on: sqlGenerator, dataSource, config
  └─ calls: sqlGenerator for SQL, executes via dataSource

SqlGenerator
  ├─ depends on: columnUtils, config
  └─ called by: client for SQL generation

// Framework: Writer.setup()
override suspend fun setup() {
    // Your code:
    namespaces.forEach { client.createNamespace(it) }
    initialStatuses = gatherer.gatherInitialStatus(names)
}

// Framework: Writer.createStreamLoader(stream)
override fun createStreamLoader(stream: DestinationStream): StreamLoader {
    // Your code:
    return when (stream.minimumGenerationId) {
        0L -> when (stream.importType) {
            is Dedupe -> DirectLoadTableDedupStreamLoader(...)
            else -> DirectLoadTableAppendStreamLoader(...)
        }
        stream.generationId -> /* truncate modes */
    }
}

// Framework: StreamLoader.start() (inside)
if (tableExists(finalTable)) {
    client.ensureSchemaMatches(stream, finalTable, columnMapping)  // Your code
} else {
    client.createTable(stream, finalTable, columnMapping, false)  // Your code
}

// Framework: Aggregate.accept(record)
override fun accept(record: RecordDTO) {
    buffer.accumulate(record.fields)  // Your code (InsertBuffer)
}

// Framework: Aggregate.flush()
override suspend fun flush() {
    buffer.flush()  // Your code (InsertBuffer writes batch)
}

// Framework: StreamLoader.close() (inside)
if (streamCompleted) {
    // Dedupe mode
    client.upsertTable(stream, columnMapping, tempTable, finalTable)  // Your code

    // Truncate mode
    client.overwriteTable(tempTable, finalTable)  // Your code
}
client.dropTable(tempTable)  // Your code

override suspend fun discoverSchema(tableName): TableSchema {
    // Query system catalog: DESCRIBE TABLE, information_schema, etc.
    // Return: Map<columnName, ColumnType(type, nullable)>
    // Filter out Airbyte metadata columns
}

override fun computeSchema(stream, columnMapping): TableSchema {
    // Map stream.schema to database types
    // Use columnUtils.toDialectType()
    // Apply column name mapping
    // Filter out Airbyte metadata columns
}

val changeset = ColumnChangeset(
    columnsToAdd = expected - actual,
    columnsToDrop = actual - expected,
    columnsToChange = actual.filter { expected[it.key] != it.value },
)

override suspend fun applyChangeset(..., changeset) {
    changeset.columnsToAdd.forEach { (name, type) ->
        execute("ALTER TABLE $table ADD COLUMN $name $type")
    }
    changeset.columnsToDrop.forEach { (name, _) ->
        execute("ALTER TABLE $table DROP COLUMN $name")
    }
    // Type changes: temp column approach or table recreation
}

-- 1. Add temp column
ALTER TABLE t ADD COLUMN col_temp VARCHAR;

-- 2. Cast and copy
UPDATE t SET col_temp = CAST(col AS VARCHAR);

-- 3. Rename original to backup
ALTER TABLE t RENAME COLUMN col TO col_backup;

-- 4. Rename temp to original
ALTER TABLE t RENAME COLUMN col_temp TO col;

-- 5. Drop backup
ALTER TABLE t DROP COLUMN col_backup;

-- 1. Create temp with new schema
CREATE TABLE temp (...) ENGINE = ReplacingMergeTree(...);

-- 2. Copy intersection
INSERT INTO temp SELECT common_columns FROM original;

-- 3. Atomic swap
EXCHANGE TABLES original AND temp;

-- 4. Drop old
DROP TABLE temp;

val hasCdc = stream.schema.asColumns().containsKey(CDC_DELETED_AT_COLUMN)
// CDC_DELETED_AT_COLUMN = "_ab_cdc_deleted_at"

MERGE INTO target
USING source
ON target.pk = source.pk
WHEN MATCHED AND source._ab_cdc_deleted_at IS NOT NULL
     AND source.cursor > target.cursor THEN DELETE
WHEN MATCHED AND source.cursor > target.cursor THEN UPDATE ...
WHEN NOT MATCHED AND source._ab_cdc_deleted_at IS NULL THEN INSERT ...

MERGE INTO target
USING source
ON target.pk = source.pk
-- No DELETE clause
WHEN MATCHED AND source.cursor > target.cursor THEN UPDATE ...
WHEN NOT MATCHED THEN INSERT ...
-- Deleted records upserted with _ab_cdc_deleted_at populated

val cdcDeleteClause = if (
    stream.schema.asColumns().containsKey(CDC_DELETED_AT_COLUMN) &&
    config.cdcDeletionMode == CdcDeletionMode.HARD_DELETE
) {
    """
    WHEN MATCHED AND new_record._ab_cdc_deleted_at IS NOT NULL
         AND $cursorComparison THEN DELETE
    """
} else {
    ""
}

val cdcSkipInsertClause = if (hasCdc && isHardDelete) {
    "AND new_record._ab_cdc_deleted_at IS NULL"
} else {
    ""
}

val mergeStatement = """
    MERGE INTO $target
    USING $source
    ON $pkMatch
    $cdcDeleteClause
    WHEN MATCHED AND $cursorComparison THEN UPDATE ...
    WHEN NOT MATCHED $cdcSkipInsertClause THEN INSERT ...
"""

data class MyConfiguration(
    val cdcDeletionMode: CdcDeletionMode = CdcDeletionMode.HARD_DELETE,
)

enum class CdcDeletionMode(@get:JsonValue val value: String) {
    HARD_DELETE("Hard delete"),
    SOFT_DELETE("Soft delete"),
}

StreamLoader.start():
  tempGenId = getGenerationId(tempTable)   // null if doesn't exist
  realGenId = getGenerationId(realTable)   // null if doesn't exist

  case 1: tempGenId == stream.generationId
    → Resume interrupted sync (write to temp)

  case 2: realGenId == stream.generationId
    → Sync already completed, STATE lost (write to real, skip finalization)

  case 3: Neither matches
    → New sync (drop stale temp if exists, create fresh temp)

"_AIRBYTE_RAW_ID" to "VARCHAR NOT NULL"
"_AIRBYTE_EXTRACTED_AT" to "TIMESTAMP_TZ NOT NULL"
"_AIRBYTE_META" to "VARIANT NOT NULL"
"_AIRBYTE_GENERATION_ID" to "NUMBER(38,0)"

"_airbyte_raw_id" to "String NOT NULL"
"_airbyte_extracted_at" to "DateTime64(3) NOT NULL"
"_airbyte_meta" to "String NOT NULL"
"_airbyte_generation_id" to "UInt32 NOT NULL"

override fun createStreamLoader(stream: DestinationStream): StreamLoader {
    val initialStatus = initialStatuses[stream]!!
    val tableNames = names[stream]!!.tableNames
    val columnMapping = names[stream]!!.columnNameMapping

    return when (stream.minimumGenerationId) {
        0L -> when (stream.importType) {
            is Dedupe -> DirectLoadTableDedupStreamLoader(
                stream, initialStatus, tableNames.finalTableName!!,
                tempTableNameGenerator.generate(tableNames.finalTableName!!),
                columnMapping, client, client, streamStateStore
            )
            else -> DirectLoadTableAppendStreamLoader(
                stream, initialStatus, tableNames.finalTableName!!,
                tempTableNameGenerator.generate(tableNames.finalTableName!!),
                columnMapping, client, client, streamStateStore
            )
        }
        stream.generationId -> when (stream.importType) {
            is Dedupe -> DirectLoadTableDedupTruncateStreamLoader(...)
            else -> DirectLoadTableAppendTruncateStreamLoader(...)
        }
        else -> throw SystemErrorException("Hybrid refresh not supported")
    }
}

fun createTable(stream: DestinationStream, tableName: TableName, ...): String {
    val columnDeclarations = stream.schema.asColumns()
        .map { (name, type) -> formatColumn(name, type) }
        .joinToString(",\n")

    return """
        CREATE TABLE ${fullyQualifiedName(tableName)} (
          ${metadataColumns}
          ${columnDeclarations}
        )
    """.trimIndent().andLog()
}

MERGE INTO final_table AS target
USING (
  SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (
      PARTITION BY primary_key
      ORDER BY _airbyte_extracted_at DESC
    ) AS rn
    FROM temp_table
  ) WHERE rn = 1
) AS source
ON target.pk = source.pk
WHEN MATCHED AND source.cursor > target.cursor THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...

-- Option 1: SWAP (if database supports)
ALTER TABLE final SWAP WITH temp;
DROP TABLE temp;

-- Option 2: EXCHANGE (ClickHouse)
EXCHANGE TABLES final AND temp;
DROP TABLE temp;

-- Option 3: DROP + RENAME
DROP TABLE IF EXISTS final;
ALTER TABLE temp RENAME TO final;

-- Add column
ALTER TABLE t ADD COLUMN new_col VARCHAR;

-- Drop column
ALTER TABLE t DROP COLUMN old_col;

-- Modify type (Postgres)
ALTER TABLE t ALTER COLUMN col TYPE VARCHAR USING col::VARCHAR;

-- Modify type (ClickHouse)
ALTER TABLE t MODIFY COLUMN col Nullable(String);

destination-{db} --spec

{
  "type": "SPEC",
  "spec": {
    "documentationUrl": "https://docs.airbyte.com/integrations/destinations/{db}",
    "connectionSpecification": {
      "$schema": "http://json-schema.org/draft-07/schema#",
      "type": "object",
      "required": ["hostname", "database", "username", "password"],
      "properties": { ... }
    },
    "supportsIncremental": true,
    "supported_destination_sync_modes": ["overwrite", "append", "append_dedup"]
  }
}

// src/test-integration/kotlin/.../spec/{DB}SpecTest.kt
class {DB}SpecTest : SpecTest()

// Validates against: src/test-integration/resources/expected-spec-oss.json

destination-{db} --check --config config.json

{
  "type": "CONNECTION_STATUS",
  "connectionStatus": {
    "status": "SUCCEEDED"
  }
}

destination-{db} --write --config config.json --catalog catalog.json < messages.jsonl

{"type":"LOG","log":{"level":"INFO","message":"Beginning sync..."}}
{"type":"LOG","log":{"level":"INFO","message":"Finished insert of 1000 rows"}}
{"type":"STATE","state":{"type":"STREAM","stream":{...},"sourceStats":{"recordCount":1000.0}}}

# Always pin to a specific version for production
cdkVersion=0.1.76

./gradlew :destination-{db}:dependencies --configuration runtimeClasspath | grep bulk-cdk

io.airbyte.bulk-cdk:bulk-cdk-core-load:0.1.76

project :airbyte-cdk:bulk:core:load

# Edit gradle.properties
cdkVersion=0.1.76  # Update to new version

./gradlew destination-{db}:upgradeCdk --cdkVersion=0.1.76

cat airbyte-cdk/bulk/version.properties