main()
  → AirbyteDestinationRunner.run(*args)
    → Parse CLI (--spec, --check, --write)
    → Create Micronaut context
    → Select Operation (SpecOperation, CheckOperation, WriteOperationV2)
    → Execute operation

WriteOperationV2.execute()
  → DestinationLifecycle.run()
    1. Writer.setup()              [Create namespaces]
    2. Initialize streams          [Create StreamLoaders]
    3. runDataPipeline()           [Process messages]
    4. Finalize streams            [MERGE/SWAP/cleanup]
    5. Teardown                    [Close connections]

Airbyte Platform          Connector Pipeline              Database
      |                          |                            |
      |-- RECORD messages ------>|                            |
      |-- STATE messages -------->|                            |
      |                          |                            |
      |                      Parse JSON                       |
      |                      Transform types                  |
      |                      Map column names                 |
      |                      Batch records                    |
      |                          |                            |
      |                    Aggregate.accept()                 |
      |                          ↓                            |
      |                  InsertBuffer.accumulate()            |
      |                          |                            |
      |                    [Buffering]                        |
      |                          |                            |
      |                    Aggregate.flush()                  |
      |                          ↓                            |
      |                  InsertBuffer.flush() --------------->| Write batch
      |                          |                            |
      |                      [Repeat]                         |
      |                          |                            |
      |                 StreamLoader.close()                  |
      |                          ↓                            |
      |                    MERGE/SWAP/nothing -------------->| Finalize
      |<----- STATE emitted -----|                            |

StreamLoader.start() {
  - Check if final table exists
  - Create/evolve final table
  - Create temp table if needed (dedupe/truncate)
  - Store target table name in state
}

[Records flow through pipeline → your InsertBuffer writes to table]

StreamLoader.close(streamCompleted) {
  if (streamCompleted) {
    // Dedupe: MERGE temp → final
    // Truncate: SWAP temp ↔ final
    // Append: nothing (already in final)
  }
  // Always cleanup temp tables
}

override fun createStreamLoader(stream: DestinationStream): StreamLoader {
    return when (stream.minimumGenerationId) {
        0L -> when (stream.importType) {
            is Dedupe -> DirectLoadTableDedupStreamLoader(...)
            else -> DirectLoadTableAppendStreamLoader(...)
        }
        stream.generationId -> when (stream.importType) {
            is Dedupe -> DirectLoadTableDedupTruncateStreamLoader(...)
            else -> DirectLoadTableAppendTruncateStreamLoader(...)
        }
        else -> throw SystemErrorException("Hybrid refresh not supported")
    }
}

class MyAggregate(private val buffer: MyInsertBuffer) : Aggregate {
    override fun accept(record: RecordDTO) {
        buffer.accumulate(record.fields)
    }

    override suspend fun flush() {
        buffer.flush()
    }
}

@Factory
class MyAggregateFactory(
    private val client: MyAirbyteClient,
    private val streamStateStore: StreamStateStore<DirectLoadTableExecutionConfig>,
) : AggregateFactory {
    override fun create(key: StoreKey): Aggregate {
        // Get table name set by StreamLoader
        val tableName = streamStateStore.get(key)!!.tableName

        val buffer = MyInsertBuffer(tableName, client)
        return MyAggregate(buffer)
    }
}

interface TableOperationsClient {
    suspend fun createNamespace(namespace: String)
    suspend fun namespaceExists(namespace: String): Boolean
    suspend fun createTable(stream, tableName, columnMapping, replace)
    suspend fun tableExists(table: TableName): Boolean
    suspend fun dropTable(tableName: TableName)
    suspend fun countTable(tableName: TableName): Long?
    suspend fun getGenerationId(tableName: TableName): Long
    suspend fun overwriteTable(source, target)  // For truncate mode
    suspend fun copyTable(columnMapping, source, target)
    suspend fun upsertTable(stream, columnMapping, source, target)  // For dedupe
}

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

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

    override suspend fun upsertTable(...) {
        execute(sqlGenerator.upsertTable(...))  // MERGE statement
    }

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

// 1. Discover current schema from database
suspend fun discoverSchema(tableName): TableSchema

// 2. Compute expected schema from stream
fun computeSchema(stream, columnMapping): TableSchema

// 3. Compare (automatic by CDK)
val changeset = ColumnChangeset(
    columnsToAdd = ...,
    columnsToDrop = ...,
    columnsToChange = ...,
)

// 4. Apply changes
suspend fun applyChangeset(..., changeset)

@Singleton
class MyWriter(
    private val names: TableCatalog,
    private val stateGatherer: DatabaseInitialStatusGatherer<DirectLoadInitialStatus>,
    private val streamStateStore: StreamStateStore<DirectLoadTableExecutionConfig>,
    private val client: MyAirbyteClient,
    private val tempTableNameGenerator: TempTableNameGenerator,
) : DestinationWriter {
    private lateinit var initialStatuses: Map<DestinationStream, DirectLoadInitialStatus>

    override suspend fun setup() {
        // Create all namespaces
        names.values
            .map { it.tableNames.finalTableName!!.namespace }
            .toSet()
            .forEach { client.createNamespace(it) }

        // Gather initial state (table exists? gen ID? columns?)
        initialStatuses = stateGatherer.gatherInitialStatus(names)
    }

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

        return /* Select appropriate StreamLoader */
    }
}

StreamLoader.start() → createTable(tempTable)
Records written to tempTable via InsertBuffer
StreamLoader.close() → finalize from tempTable → dropTable(tempTable)

WITH deduped AS (
  SELECT *, ROW_NUMBER() OVER (
    PARTITION BY primary_key
    ORDER BY _airbyte_extracted_at DESC
  ) AS rn
  FROM temp_table
)
SELECT * FROM deduped WHERE rn = 1

MERGE INTO final_table AS target
USING deduped AS source
ON target.pk = source.pk
WHEN MATCHED AND source.cursor > target.cursor THEN UPDATE ...
WHEN NOT MATCHED THEN INSERT ...

MERGE INTO final_table AS target
USING deduped AS source
ON target.pk = source.pk
WHEN MATCHED AND source._ab_cdc_deleted_at IS NOT NULL THEN DELETE  -- Hard delete
WHEN MATCHED AND source.cursor > target.cursor THEN UPDATE ...
WHEN NOT MATCHED AND source._ab_cdc_deleted_at IS NULL THEN INSERT ...

RECORD messages → buffer
STATE message → flush buffers → database COMMIT → emit STATE

Pipeline receives: JSON message
  ↓ Deserialize
AirbyteValue (StringValue, IntegerValue, etc.)
  ↓ Transform
Database-specific format (via ColumnUtils)
  ↓ Buffer
InsertBuffer accumulates
  ↓ Format
CSV, binary, JSON, etc.
  ↓ Write
Database client writes batch

// Source
{"id": 123, "name": "Alice", "created_at": "2024-01-01T12:00:00Z"}

// After transformation (RecordDTO.fields)
{
  "id": IntegerValue(123),
  "name": StringValue("Alice"),
  "created_at": TimestampValue("2024-01-01T12:00:00Z"),
  "_airbyte_raw_id": StringValue("uuid..."),
  "_airbyte_extracted_at": TimestampValue("2024-01-01T12:00:00Z"),
  "_airbyte_meta": ObjectValue(...),
  "_airbyte_generation_id": IntegerValue(42)
}

// InsertBuffer formats for database
CSV: "123,Alice,2024-01-01 12:00:00,uuid...,2024-01-01 12:00:00,{},42"
Binary: [0x7B, 0x00, ...] (database-specific format)

Stream schema:     {"field_name": StringType}
                          ↓
ColumnNameGenerator:  "field_name" → "FIELD_NAME" (Snowflake)
                                   → "field_name" (ClickHouse)
                          ↓
TableCatalog stores:  {"field_name": "FIELD_NAME"}
                          ↓
Your code queries:    columnMapping["field_name"] → "FIELD_NAME"

try {
    connection.executeQuery(sql)
} catch (e: SQLException) {
    when {
        e.message?.contains("permission") == true ->
            throw ConfigErrorException("Permission denied. Grant privileges.", e)
        e.message?.contains("timeout") == true ->
            throw TransientErrorException("Network timeout. Will retry.", e)
        else ->
            throw SystemErrorException("Unexpected SQL error", e)
    }
}

StreamLoader.close(streamCompleted = false) {
    // Skip finalization (no MERGE/SWAP)
    // Drop temp tables (cleanup)
    // Final table unchanged
}

-- Dedupe in temp
CREATE TABLE deduped AS SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY pk ORDER BY cursor DESC) AS rn
  FROM temp
) WHERE rn = 1;

-- DELETE + INSERT
DELETE FROM final WHERE pk IN (SELECT pk FROM deduped);
INSERT INTO final SELECT * FROM deduped;

-- Update existing
UPDATE final SET col1 = temp.col1, ... FROM temp WHERE final.pk = temp.pk;

-- Insert new
INSERT INTO final SELECT * FROM temp WHERE pk NOT IN (SELECT pk FROM final);

-- Postgres: INSERT ... ON CONFLICT
INSERT INTO final SELECT * FROM temp
ON CONFLICT (pk) DO UPDATE SET col1 = EXCLUDED.col1, ...;

-- MySQL: INSERT ... ON DUPLICATE KEY UPDATE
INSERT INTO final SELECT * FROM temp
ON DUPLICATE KEY UPDATE col1 = VALUES(col1), ...;

cdkVersion=0.1.76  # Pin to specific version

cdkVersion=local  # Only for CDK development

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

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