#!/usr/bin/env bash

set -Eeuo pipefail

# CI usage:
#   bash e2e-test/run_local_e2e.sh
#
# Common overrides:
#   PYTHON_VENV=.venv SKIP_VENV=1 TIDB_VERSION=v8.5.6 TIUP_TAG=tidb-fivetran

fail() {
    printf 'ERROR: %s\n' "$*" >&2
    exit 1
}

log() {
    printf '%s\n' "$*"
}

run() {
    printf '+'
    printf ' %q' "$@"
    printf '\n'
    "$@" || fail "command failed ($?): $*"
}

require_command() {
    command -v "$1" >/dev/null 2>&1 || fail "missing required command: $1"
}

script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)" || exit 1
repo_root="$(cd "${script_dir}/.." && pwd -P)" || exit 1
cd "$repo_root" || exit 1

: "${PYTHON_VENV:=.venv}"
: "${SKIP_VENV:=0}"
: "${TIDB_VERSION:=v8.5.6}"
: "${TIUP_TAG:=tidb-fivetran}"
: "${CHANGEFEED_ID:=tidb-fivetran-local-e2e}"
: "${DB_NAME:=test}"
: "${TABLE_NAME:=fruit1}"
: "${DUCKDB_TABLE:=tester.test_fruit_1}"
: "${WORKSPACE:=examples/local_workspace}"
: "${CONFIG_PATH:=configuration.json}"
: "${WAREHOUSE:=files/warehouse.db}"
: "${E2E_RUNTIME_DIR:=e2e-test/.tmp/local-e2e}"
: "${TEST_SQL_PATH:=e2e-test/test.sql}"
: "${TIDB_HOST:=127.0.0.1}"
: "${TIDB_PORT:=4000}"
: "${TIDB_USER:=root}"
: "${TIDB_PASSWORD:=}"
: "${TICDC_SERVER:=127.0.0.1:8300}"
: "${CDC_CLI:=./bin/cdc}"
: "${DUMPLING_THREADS:=8}"
: "${DUMPLING_ROWS:=200000}"
: "${DUMPLING_FILE_SIZE:=64MiB}"
: "${READY_TIMEOUT_SECONDS:=180}"
: "${FIVETRAN_DEBUG_TIMEOUT_SECONDS:=180}"
: "${PLAYGROUND_LOG_TAIL_LINES:=200}"
: "${KEEP_PLAYGROUND:=0}"
: "${TIMEOUT_BIN:=timeout}"

tag="$TIUP_TAG"
db_name="$DB_NAME"
table_name="$TABLE_NAME"
duckdb_table="$DUCKDB_TABLE"
changefeed_id="$CHANGEFEED_ID"
workspace="$WORKSPACE"
snapshot_dir="$workspace/snapshot"
increment_dir="$workspace/increment"
config_path="$CONFIG_PATH"
warehouse="$WAREHOUSE"
runtime_dir="$E2E_RUNTIME_DIR"
test_sql_path="$TEST_SQL_PATH"
playground_log="$runtime_dir/playground.log"
ticdc_config="$runtime_dir/ticdc.toml"
changefeed_config="$runtime_dir/changefeed.toml"
tiup_data_dir="$HOME/.tiup/data/$tag"
created_config=0
playground_pid=
cdc_cli=("$CDC_CLI" cli)

kill_tagged_tiup_processes() {
    command -v pgrep >/dev/null 2>&1 || return 0

    local pids
    pids=$(pgrep -f "$tiup_data_dir/" 2>/dev/null || true)
    if [ -z "$pids" ]; then
        return 0
    fi

    # shellcheck disable=SC2086
    kill -TERM $pids >/dev/null 2>&1 || true
    sleep 2
    pids=$(pgrep -f "$tiup_data_dir/" 2>/dev/null || true)
    if [ -n "$pids" ]; then
        # shellcheck disable=SC2086
        kill -KILL $pids >/dev/null 2>&1 || true
    fi
}

cleanup() {
    if [ -n "${playground_pid:-}" ]; then
        if [ "$KEEP_PLAYGROUND" = "1" ]; then
            log "Leaving playground running with pid $playground_pid because KEEP_PLAYGROUND=1"
        else
            kill -TERM -- -"$playground_pid" >/dev/null 2>&1 || true
            kill -TERM -- "$playground_pid" >/dev/null 2>&1 || true
            sleep 2
            kill -KILL -- -"$playground_pid" >/dev/null 2>&1 || true
            kill -KILL -- "$playground_pid" >/dev/null 2>&1 || true
            wait "$playground_pid" 2>/dev/null || true
        fi
    fi

    if [ "$created_config" = "1" ]; then
        rm -f "$config_path"
    fi

    rm -f "$ticdc_config" "$changefeed_config"

    if [ "$KEEP_PLAYGROUND" != "1" ]; then
        tiup clean "$tag" >/dev/null 2>&1 || true
        kill_tagged_tiup_processes
    fi
}

on_exit() {
    status=$?
    if [ "$status" -ne 0 ] && [ -f "$playground_log" ]; then
        {
            printf '\n===== playground log tail (%s lines) =====\n' "$PLAYGROUND_LOG_TAIL_LINES"
            tail -n "$PLAYGROUND_LOG_TAIL_LINES" "$playground_log" || true
            printf '===== end playground log tail =====\n'
        } >&2
    fi
    cleanup
    exit "$status"
}

trap on_exit EXIT

activate_python() {
    if [ "$SKIP_VENV" = "1" ] || [ -n "${VIRTUAL_ENV:-}" ]; then
        return 0
    fi

    if [ -f "${PYTHON_VENV}/bin/activate" ]; then
        source "${PYTHON_VENV}/bin/activate"
        return 0
    fi

    if command -v fivetran >/dev/null 2>&1; then
        log "No ${PYTHON_VENV}/bin/activate found; using current Python environment."
        return 0
    fi

    fail "missing ${PYTHON_VENV}/bin/activate and fivetran is not on PATH"
}

validate_identifier() {
    local value=$1
    local label=$2

    if [[ ! "$value" =~ ^[A-Za-z_][A-Za-z0-9_]*$ ]]; then
        fail "invalid $label: $value"
    fi
}

wait_until() {
    local description=$1
    local timeout_seconds=$2
    shift 2

    local deadline=$((SECONDS + timeout_seconds))
    while [ "$SECONDS" -lt "$deadline" ]; do
        if "$@"; then
            return 0
        fi
        sleep 2
    done

    fail "$description did not become ready within ${timeout_seconds}s"
}

tidb_ready() {
    mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u "$TIDB_USER" --password="$TIDB_PASSWORD" -e "select 1" >/dev/null 2>&1
}

ticdc_ready() {
    "${cdc_cli[@]}" changefeed list --server="$TICDC_SERVER" >/dev/null 2>&1
}

mysql_exec() {
    local sql=$1

    mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u "$TIDB_USER" --password="$TIDB_PASSWORD" -e "$sql" \
        || fail "mysql failed: $sql"
}

mysql_try_exec() {
    local sql=$1

    if ! mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u "$TIDB_USER" --password="$TIDB_PASSWORD" -e "$sql"; then
        log "Optional TiDB setting failed and was ignored: $sql"
    fi
}

mysql_exec_file() {
    local file=$1

    [ -f "$file" ] || fail "missing SQL file: $file"
    mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u "$TIDB_USER" --password="$TIDB_PASSWORD" <"$file" \
        || fail "mysql failed while executing $file"
}

has_file_matching() {
    local dir=$1
    local pattern=$2

    find "$dir" -type f -name "$pattern" -print -quit 2>/dev/null | grep -q .
}

wait_for_increment_outputs() {
    local deadline=$((SECONDS + READY_TIMEOUT_SECONDS))
    while [ "$SECONDS" -lt "$deadline" ]; do
        if [ -f "$increment_dir/metadata" ] &&
            has_file_matching "$increment_dir" 'CDC*.csv' &&
            has_file_matching "$increment_dir" 'CDC.index' &&
            has_file_matching "$increment_dir" 'schema_*.json'; then
            return 0
        fi
        sleep 2
    done

    fail "TiCDC did not write expected increment files under $increment_dir"
}

metadata_checkpoint_ts() {
    [ -f "$increment_dir/metadata" ] || return 1
    sed -n 's/.*"checkpoint-ts"[[:space:]]*:[[:space:]]*"\{0,1\}\([0-9][0-9]*\)"\{0,1\}.*/\1/p' "$increment_dir/metadata" | tail -n 1
}

increment_row_commit_ts() {
    local needle=$1

    find "$increment_dir" -type f -name 'CDC*.csv' -exec grep -h -F "$needle" {} + 2>/dev/null |
        awk -F',' 'NR == 1 {print $4; exit}'
}

final_ddl_row_checkpoint_ready() {
    local row_commit_ts
    local checkpoint_ts

    row_commit_ts=$(increment_row_commit_ts "fruit-after-truncate" || true)
    checkpoint_ts=$(metadata_checkpoint_ts || true)
    if [[ ! "$row_commit_ts" =~ ^[0-9]+$ ]] || [[ ! "$checkpoint_ts" =~ ^[0-9]+$ ]]; then
        return 1
    fi

    [ "$checkpoint_ts" -ge "$row_commit_ts" ]
}

test_sql_final_row_checkpoint_ready() {
    local row_commit_ts
    local checkpoint_ts

    row_commit_ts=$(increment_row_commit_ts "cross-family value after ddl" || true)
    checkpoint_ts=$(metadata_checkpoint_ts || true)
    if [[ ! "$row_commit_ts" =~ ^[0-9]+$ ]] || [[ ! "$checkpoint_ts" =~ ^[0-9]+$ ]]; then
        return 1
    fi

    [ "$checkpoint_ts" -ge "$row_commit_ts" ]
}

wait_for_duckdb_table() {
    wait_for_duckdb_relation "$duckdb_table"
}

wait_for_duckdb_relation() {
    local relation=$1
    local deadline=$((SECONDS + 30))
    while [ "$SECONDS" -lt "$deadline" ]; do
        if duckdb "$warehouse" "select 1 from $relation limit 1;" >/dev/null 2>&1; then
            return 0
        fi
        sleep 1
    done

    log "DuckDB tables:"
    duckdb "$warehouse" "show tables;" || true
    fail "DuckDB relation $relation was not created"
}

verify_duckdb_rows() {
    local row_diff_count
    local metadata_mismatch_count
    local active_row_count
    local pre_truncate_deleted_count
    local no_pk_table_count

    row_diff_count=$(duckdb -noheader -csv "$warehouse" "
with expected_rows(id, name, color, sweetness, deleted) as (
    values
        (1, 'apple', null, null, true),
        (2, 'banana', null, null, true),
        (3, 'cherry', null, null, true),
        (4, 'date', null, null, true),
        (5, 'elderberry', null, null, true),
        (6, 'fig', null, null, true),
        (7, 'grape', null, null, true),
        (8, 'honeydew', null, null, true),
        (9, 'kiwi', null, null, true),
        (10, 'lemon', null, null, true),
        (11, 'fruit-11', null, null, true),
        (12, 'fruit-12', 'red', null, true),
        (13, 'fruit-13', 'yellow', 7, true),
        (14, 'fruit-14-long-name-after-widen', 'purple', 8, true),
        (100, 'fruit-after-truncate', 'green', 9, false)
),
actual_rows as (
    select id, name, color, sweetness, coalesce(_fivetran_deleted, false) as deleted
    from $duckdb_table
),
missing_expected as (
    select id, name, color, sweetness, deleted from expected_rows
    except all
    select id, name, color, sweetness, deleted from actual_rows
),
unexpected_actual as (
    select id, name, color, sweetness, deleted from actual_rows
    except all
    select id, name, color, sweetness, deleted from expected_rows
)
select
    (select count(*) from missing_expected) +
    (select count(*) from unexpected_actual);
" | tr -d '[:space:]')

    if [ -z "$row_diff_count" ] || [ "$row_diff_count" -ne 0 ]; then
        log "DuckDB row data mismatch. Missing expected rows:"
        duckdb "$warehouse" "
with expected_rows(id, name, color, sweetness, deleted) as (
    values
        (1, 'apple', null, null, true),
        (2, 'banana', null, null, true),
        (3, 'cherry', null, null, true),
        (4, 'date', null, null, true),
        (5, 'elderberry', null, null, true),
        (6, 'fig', null, null, true),
        (7, 'grape', null, null, true),
        (8, 'honeydew', null, null, true),
        (9, 'kiwi', null, null, true),
        (10, 'lemon', null, null, true),
        (11, 'fruit-11', null, null, true),
        (12, 'fruit-12', 'red', null, true),
        (13, 'fruit-13', 'yellow', 7, true),
        (14, 'fruit-14-long-name-after-widen', 'purple', 8, true),
        (100, 'fruit-after-truncate', 'green', 9, false)
),
actual_rows as (
    select id, name, color, sweetness, coalesce(_fivetran_deleted, false) as deleted
    from $duckdb_table
)
select id, name, color, sweetness, deleted from expected_rows
except all
select id, name, color, sweetness, deleted from actual_rows
order by id;
"
        log "Unexpected actual rows:"
        duckdb "$warehouse" "
with expected_rows(id, name, color, sweetness, deleted) as (
    values
        (1, 'apple', null, null, true),
        (2, 'banana', null, null, true),
        (3, 'cherry', null, null, true),
        (4, 'date', null, null, true),
        (5, 'elderberry', null, null, true),
        (6, 'fig', null, null, true),
        (7, 'grape', null, null, true),
        (8, 'honeydew', null, null, true),
        (9, 'kiwi', null, null, true),
        (10, 'lemon', null, null, true),
        (11, 'fruit-11', null, null, true),
        (12, 'fruit-12', 'red', null, true),
        (13, 'fruit-13', 'yellow', 7, true),
        (14, 'fruit-14-long-name-after-widen', 'purple', 8, true),
        (100, 'fruit-after-truncate', 'green', 9, false)
),
actual_rows as (
    select id, name, color, sweetness, coalesce(_fivetran_deleted, false) as deleted
    from $duckdb_table
)
select id, name, color, sweetness, deleted from actual_rows
except all
select id, name, color, sweetness, deleted from expected_rows
order by id;
"
        fail "DuckDB source rows do not exactly match expected rows"
    fi

    active_row_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from $duckdb_table
where coalesce(_fivetran_deleted, false) = false;
" | tr -d '[:space:]')

    if [ -z "$active_row_count" ] || [ "$active_row_count" -ne 1 ]; then
        fail "expected exactly one active post-truncate row in $duckdb_table, got $active_row_count"
    fi

    pre_truncate_deleted_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from $duckdb_table
where id between 1 and 14 and coalesce(_fivetran_deleted, false) = true;
" | tr -d '[:space:]')

    if [ -z "$pre_truncate_deleted_count" ] || [ "$pre_truncate_deleted_count" -ne 14 ]; then
        fail "expected 14 soft-deleted pre-truncate rows in $duckdb_table, got $pre_truncate_deleted_count"
    fi

    no_pk_table_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from information_schema.tables
where table_schema = 'tester' and table_name = 'test_ddl_no_pk';
" | tr -d '[:space:]')

    if [ -z "$no_pk_table_count" ] || [ "$no_pk_table_count" -ne 0 ]; then
        fail "expected no DuckDB table for source table without primary key, got $no_pk_table_count"
    fi

    metadata_mismatch_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from $duckdb_table
where
    _incremental_source_schema <> '$db_name'
    or _incremental_source_table <> '$table_name'
    or (id between 1 and 10 and coalesce(_incremental_commit_ts, -1) <> 0)
    or (id in (11, 12, 13, 14, 100) and coalesce(_incremental_commit_ts, 0) <= 0);
" | tr -d '[:space:]')

    if [ -z "$metadata_mismatch_count" ] || [ "$metadata_mismatch_count" -ne 0 ]; then
        duckdb "$warehouse" "
select id, name, color, sweetness, _incremental_commit_ts, _incremental_source_schema, _incremental_source_table, _fivetran_deleted
from $duckdb_table
where
    _incremental_source_schema <> '$db_name'
    or _incremental_source_table <> '$table_name'
    or (id between 1 and 10 and coalesce(_incremental_commit_ts, -1) <> 0)
    or (id in (11, 12, 13, 14, 100) and coalesce(_incremental_commit_ts, 0) <= 0)
order by id;
"
        fail "DuckDB stable metadata columns do not match expected values"
    fi
}

verify_test_sql_duckdb_results() {
    local matrix_table="tester.ft_ddl_type_test_ft_ddl_type_matrix"
    local create_table="tester.ft_ddl_type_test_ft_ddl_create_after_snapshot"
    local truncate_table="tester.ft_ddl_type_test_ft_ddl_truncate_target"
    local matrix_diff_count
    local deleted_id_two_count
    local create_row_count
    local truncate_active_count
    local truncate_deleted_count
    local no_stable_pk_table_count

    matrix_diff_count=$(duckdb -noheader -csv "$warehouse" "
with expected_matrix_rows(
    id,
    c_varchar,
    c_added_nullable,
    c_default_toggle,
    c_changed_rename_type,
    c_type_cross_family_number
) as (
    values
        (1, 'after ddl update', 'after ddl update non-null', 'explicit after default drop', 9001, 'number column changed to varchar'),
        (3, null, null, 'v1', null, null),
        (4, null, null, 'v2', null, null),
        (5, null, 'insert after ddl', 'explicit_insert_v5', 9005, 'cross-family value after ddl'),
        (6, null, null, null, null, null),
        (7, null, null, null, null, null),
        (8, null, null, null, null, null)
),
actual_matrix_rows as (
    select
        id,
        c_varchar,
        c_added_nullable,
        c_default_toggle,
        c_changed_rename_type,
        c_type_cross_family_number
    from $matrix_table
    where coalesce(_fivetran_deleted, false) = false
),
missing_expected as (
    select * from expected_matrix_rows
    except all
    select * from actual_matrix_rows
),
unexpected_actual as (
    select * from actual_matrix_rows
    except all
    select * from expected_matrix_rows
)
select
    (select count(*) from missing_expected) +
    (select count(*) from unexpected_actual);
" | tr -d '[:space:]')

    if [ -z "$matrix_diff_count" ] || [ "$matrix_diff_count" -ne 0 ]; then
        log "DuckDB ft_ddl_type_matrix active row mismatch. Active rows:"
        duckdb "$warehouse" "
select
    id,
    c_varchar,
    c_added_nullable,
    c_default_toggle,
    c_changed_rename_type,
    c_type_cross_family_number,
    _fivetran_deleted
from $matrix_table
order by id;
"
        fail "DuckDB ft_ddl_type_matrix active rows do not match test.sql expectations"
    fi

    deleted_id_two_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from $matrix_table
where id = 2 and coalesce(_fivetran_deleted, false) = true;
" | tr -d '[:space:]')

    if [ -z "$deleted_id_two_count" ] || [ "$deleted_id_two_count" -ne 1 ]; then
        fail "expected ft_ddl_type_matrix id=2 to be soft deleted, got $deleted_id_two_count"
    fi

    create_row_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from $create_table
where
    id = 1
    and payload = 'new table with stable primary key'
    and coalesce(_fivetran_deleted, false) = false;
" | tr -d '[:space:]')

    if [ -z "$create_row_count" ] || [ "$create_row_count" -ne 1 ]; then
        fail "expected ft_ddl_create_after_snapshot row to sync, got $create_row_count"
    fi

    truncate_active_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from $truncate_table
where id = 3 and payload = 'after truncate' and coalesce(_fivetran_deleted, false) = false;
" | tr -d '[:space:]')

    truncate_deleted_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from $truncate_table
where id in (1, 2) and coalesce(_fivetran_deleted, false) = true;
" | tr -d '[:space:]')

    if [ -z "$truncate_active_count" ] || [ "$truncate_active_count" -ne 1 ] ||
        [ -z "$truncate_deleted_count" ] || [ "$truncate_deleted_count" -ne 2 ]; then
        fail "expected ft_ddl_truncate_target truncate semantics, active=$truncate_active_count deleted=$truncate_deleted_count"
    fi

    no_stable_pk_table_count=$(duckdb -noheader -csv "$warehouse" "
select count(*)
from information_schema.tables
where table_schema = 'tester' and table_name = 'ft_ddl_type_test_ft_ddl_no_stable_pk';
" | tr -d '[:space:]')

    if [ -z "$no_stable_pk_table_count" ] || [ "$no_stable_pk_table_count" -ne 0 ]; then
        fail "expected no DuckDB table for ft_ddl_no_stable_pk, got $no_stable_pk_table_count"
    fi
}

run_fivetran_debug_cycle() {
    local phase=$1
    local code=0

    log "+ fivetran debug --configuration $config_path ($phase)"
    set +e
    "$TIMEOUT_BIN" "${FIVETRAN_DEBUG_TIMEOUT_SECONDS}s" fivetran debug --configuration "$config_path"
    code=$?
    set -e

    if [ "$code" -eq 124 ]; then
        log "fivetran debug timed out after ${FIVETRAN_DEBUG_TIMEOUT_SECONDS}s during $phase; continuing to verification"
    elif [ "$code" -ne 0 ]; then
        fail "fivetran debug failed during $phase with exit code $code"
    fi
}

write_runtime_configs() {
    mkdir -p "$runtime_dir" || fail "failed to prepare $runtime_dir"

    cat >"$ticdc_config" <<'EOF'
newarch = true
EOF

    cat >"$changefeed_config" <<'EOF'
[sink]
[sink.csv]
include-commit-ts = true
EOF
}

resolve_cdc_cli() {
    if [ -x "$CDC_CLI" ]; then
        cdc_cli=("$CDC_CLI" cli)
        return 0
    fi

    cdc_cli=(tiup "ctl:${TIDB_VERSION}" cdc)
}

prepare_test_sql_environment() {
    mysql_try_exec "set global tidb_enable_check_constraint = on;"
    mysql_try_exec "set global tidb_enable_foreign_key = on;"
}

apply_ddl_test_cases() {
    local ddl_sql

    ddl_sql="
use $db_name;
insert into $table_name(name) values ('fruit-11');
alter table $table_name add column color varchar(20) default null;
insert into $table_name(name, color) values ('fruit-12', 'red');
alter table $table_name add column sweetness int not null default 0;
insert into $table_name(name, color, sweetness) values ('fruit-13', 'yellow', 7);
alter table $table_name modify column name varchar(64) not null default '';
insert into $table_name(name, color, sweetness) values ('fruit-14-long-name-after-widen', 'purple', 8);
alter table $table_name comment = 'local e2e ddl coverage';
create table ddl_no_pk(note varchar(20));
insert into ddl_no_pk(note) values ('ignored');
truncate table $table_name;
insert into $table_name(id, name, color, sweetness) values (100, 'fruit-after-truncate', 'green', 9);
"
    mysql_exec "$ddl_sql"
}

activate_python

for tool in tiup mysql fivetran duckdb "$TIMEOUT_BIN" find realpath grep tr sed awk; do
    require_command "$tool"
done

[ -f examples/local_configuration.json ] || fail "missing examples/local_configuration.json"
[ -f "$test_sql_path" ] || fail "missing $test_sql_path"
validate_identifier "$db_name" DB_NAME
validate_identifier "$table_name" TABLE_NAME

if [ ! -f "$config_path" ]; then
    cp examples/local_configuration.json "$config_path" || fail "failed to create $config_path"
    created_config=1
fi

rm -rf "$snapshot_dir" "$increment_dir" "$warehouse" files/state.json "$runtime_dir"
mkdir -p "$workspace" files || fail "failed to prepare local workspace"
write_runtime_configs
resolve_cdc_cli

tiup clean "$tag" >/dev/null 2>&1 || true

log "+ tiup playground $TIDB_VERSION --tag $tag --db 1 --pd 1 --kv 1 --tiflash 0 --ticdc 1 --ticdc.config $ticdc_config"
tiup playground "$TIDB_VERSION" --tag "$tag" --db 1 --pd 1 --kv 1 --tiflash 0 --ticdc 1 --ticdc.config "$ticdc_config" >"$playground_log" 2>&1 &
playground_pid=$!

wait_until "TiDB" "$READY_TIMEOUT_SECONDS" tidb_ready
wait_until "TiCDC" "$READY_TIMEOUT_SECONDS" ticdc_ready

seed_sql="drop database if exists $db_name; create database $db_name; use $db_name; create table fruit1(id bigint not null auto_increment primary key, name varchar(20) not null default ''); insert into fruit1(name) values ('apple'), ('banana'), ('cherry'), ('date'), ('elderberry'), ('fig'), ('grape'), ('honeydew'), ('kiwi'), ('lemon');"
mysql_exec "$seed_sql"

dumpling_cmd=(
    tiup dumpling
    -u "$TIDB_USER"
    -P "$TIDB_PORT"
    -h "$TIDB_HOST"
    --filetype csv
    -t "$DUMPLING_THREADS"
    -r "$DUMPLING_ROWS"
    -F "$DUMPLING_FILE_SIZE"
    -B "$db_name"
    -o "./$snapshot_dir"
)
if [ -n "$TIDB_PASSWORD" ]; then
    dumpling_cmd+=(-p "$TIDB_PASSWORD")
fi
run "${dumpling_cmd[@]}"

if [ ! -f "$snapshot_dir/metadata" ] || ! has_file_matching "$snapshot_dir" "$db_name.$table_name.*.csv"; then
    fail "Dumpling did not create expected CSV snapshot files under $snapshot_dir"
fi

run_fivetran_debug_cycle snapshot

increment_abs=$(realpath "$increment_dir") || fail "failed to resolve $increment_dir"
sink_uri="file://$increment_abs?protocol=csv&output-old-value=true"
# The ctl help describes this operation as "cdc cli changefeed create".
run "${cdc_cli[@]}" changefeed create --server="$TICDC_SERVER" --sink-uri="$sink_uri" --changefeed-id="$changefeed_id" --config="$changefeed_config" --no-confirm

apply_ddl_test_cases
prepare_test_sql_environment
mysql_exec_file "$test_sql_path"
wait_for_increment_outputs
wait_until "TiCDC final DDL row checkpoint" "$READY_TIMEOUT_SECONDS" final_ddl_row_checkpoint_ready
wait_until "TiCDC test.sql final row checkpoint" "$READY_TIMEOUT_SECONDS" test_sql_final_row_checkpoint_ready

run_fivetran_debug_cycle incremental
[ -f "$warehouse" ] || fail "missing DuckDB warehouse: $warehouse"
wait_for_duckdb_table
wait_for_duckdb_relation "tester.ft_ddl_type_test_ft_ddl_type_matrix"
wait_for_duckdb_relation "tester.ft_ddl_type_test_ft_ddl_create_after_snapshot"
wait_for_duckdb_relation "tester.ft_ddl_type_test_ft_ddl_truncate_target"

row_count=$(duckdb -noheader -csv "$warehouse" "select count(*) from $duckdb_table;" | tr -d '[:space:]')
incremental_count=$(duckdb -noheader -csv "$warehouse" "select count(*) from $duckdb_table where name = 'fruit-after-truncate' and coalesce(_fivetran_deleted, false) = false;" | tr -d '[:space:]')

if [ -z "$row_count" ] || [ "$row_count" -ne 15 ]; then
    fail "expected exactly 15 rows in $duckdb_table after DDL coverage, got $row_count"
fi

if [ -z "$incremental_count" ] || [ "$incremental_count" -lt 1 ]; then
    fail "expected active post-truncate row fruit-after-truncate in $duckdb_table"
fi

verify_duckdb_rows
verify_test_sql_duckdb_results

log "OK: DuckDB $warehouse contains exactly matching DDL lifecycle rows and test.sql matrix results"