# Snowflake E2E Manual Run Guide

This guide describes how to manually run the Snowflake E2E flow that mirrors
`e2e-test/run_snowflake_e2e.sh`.

The commands below assume `bash`. If your login shell is `fish`, run `bash`
from the repository root first, or translate `export` and `read` commands to
`set -x`.

## 1. Prepare Local Tools

Run from the repository root:

```bash
cd /Users/zhongtenghui/go/src/github.com/tidbcloud/tidb-fivetran-connector
UV_HTTP_TIMEOUT=120 uv sync
source .venv/bin/activate
command -v fivetran
command -v tiup
command -v mysql
command -v aws
```

If `command -v fivetran` prints nothing, `fivetran-connector-sdk` is not
installed in the active environment. Re-run `UV_HTTP_TIMEOUT=120 uv sync` and
source `.venv/bin/activate` again.

## 2. Prepare Run Variables

Use a unique run ID so the Fivetran state and S3 prefix do not collide with a
previous run.

```bash
export RUN_ID="tidb_fivetran_snowflake_e2e_$(date -u +%Y%m%d%H%M%S)"
export TIUP_TAG="tidb-fivetran-snowflake"
export TIDB_VERSION="v8.5.6"
export DB_NAME="test"
export TABLE_NAME="fruit1"
export TIDB_HOST="127.0.0.1"
export TIDB_PORT="4000"
export TICDC_SERVER="127.0.0.1:8300"
export WORKSPACE="examples/snowflake_workspace"
export RUNTIME_DIR="e2e-test/.tmp/snowflake-e2e-manual"
export S3_BASE_PREFIX="tidb-fivetran-e2e"
export S3_RUN_PREFIX="${S3_BASE_PREFIX}/${RUN_ID}"
export FIVETRAN_CONNECTION_NAME="${RUN_ID}"
export SYNC_TASK_ID="${RUN_ID}"
export SNOWFLAKE_SCHEMA="${FIVETRAN_CONNECTION_NAME}"
```

## 3. Enter Credentials

Do not reuse your local development AWS environment. Clear it first, then type
the S3 key pair that should be used for this E2E run.

```bash
unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN AWS_PROFILE

read -r -p "Fivetran API key base64: " FIVETRAN_API_KEY_BASE64
read -r -p "Fivetran destination name: " FIVETRAN_DESTINATION_NAME

read -r -p "S3 bucket: " S3_BUCKET
read -r -p "S3 region: " S3_REGION
read -r -p "S3 access key: " S3_AK
read -r -s -p "S3 secret key: " S3_SK; echo
read -r -p "S3 session token, optional: " S3_SESSION_TOKEN

read -r -p "Snowflake account: " SNOWFLAKE_ACCOUNT
read -r -p "Snowflake user: " SNOWFLAKE_USER
read -r -s -p "Snowflake password: " SNOWFLAKE_PASSWORD; echo
read -r -p "Snowflake warehouse: " SNOWFLAKE_WAREHOUSE
read -r -p "Snowflake database: " SNOWFLAKE_DATABASE
read -r -p "Snowflake role, optional: " SNOWFLAKE_ROLE

export FIVETRAN_API_KEY_BASE64 FIVETRAN_DESTINATION_NAME
export S3_BUCKET S3_REGION S3_AK S3_SK S3_SESSION_TOKEN
export SNOWFLAKE_ACCOUNT SNOWFLAKE_USER SNOWFLAKE_PASSWORD
export SNOWFLAKE_WAREHOUSE SNOWFLAKE_DATABASE SNOWFLAKE_ROLE SNOWFLAKE_SCHEMA
```

If you have raw Fivetran API key and secret instead of the base64 value:

```bash
read -r -p "Fivetran API key: " FIVETRAN_API_KEY
read -r -s -p "Fivetran API secret: " FIVETRAN_API_SECRET; echo
export FIVETRAN_API_KEY_BASE64="$(printf '%s:%s' "$FIVETRAN_API_KEY" "$FIVETRAN_API_SECRET" | base64 | tr -d '\n')"
```

Configure AWS CLI only from the S3 credentials you just typed:

```bash
export AWS_ACCESS_KEY_ID="$S3_AK"
export AWS_SECRET_ACCESS_KEY="$S3_SK"
export AWS_DEFAULT_REGION="$S3_REGION"
if [ -n "$S3_SESSION_TOKEN" ]; then
  export AWS_SESSION_TOKEN="$S3_SESSION_TOKEN"
else
  unset AWS_SESSION_TOKEN
fi

AWS_S3=(aws)

aws sts get-caller-identity
```

The automated script prints a `Runtime configuration:` summary after collecting
inputs. Non-sensitive values are printed directly. Secrets are shown as
`<redacted len=N tail=xxxx>` so you can verify which value was loaded without
leaking the full credential. Set `SHOW_RUNTIME_CONFIG=0` to suppress this
summary.

## 4. Start TiDB And TiCDC

```bash
rm -rf "$RUNTIME_DIR" "$WORKSPACE/snapshot" "$WORKSPACE/increment"
mkdir -p "$RUNTIME_DIR" "$WORKSPACE"

cat > "$RUNTIME_DIR/ticdc.toml" <<'EOF'
newarch = true
EOF

tiup clean "$TIUP_TAG" >/dev/null 2>&1 || true
TICDC_HOST="${TICDC_SERVER%:*}"
TICDC_PORT="${TICDC_SERVER##*:}"
tiup playground "$TIDB_VERSION" \
  --tag "$TIUP_TAG" \
  --host "$TIDB_HOST" \
  --db 1 \
  --db.port "$TIDB_PORT" \
  --pd 1 \
  --kv 1 \
  --tiflash 0 \
  --ticdc 1 \
  --ticdc.host "$TICDC_HOST" \
  --ticdc.port "$TICDC_PORT" \
  --ticdc.config "$RUNTIME_DIR/ticdc.toml" \
  > "$RUNTIME_DIR/playground.log" 2>&1 &
export PLAYGROUND_PID=$!
```

Wait until TiDB is ready:

```bash
until mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root -e "select 1" >/dev/null 2>&1; do
  sleep 2
done
```

Resolve the CDC CLI. Prefer `./bin/cdc` if it exists; otherwise use TiUP:

```bash
if [ -x ./bin/cdc ]; then
  CDC_CMD=(./bin/cdc cli)
else
  CDC_CMD=(tiup "ctl:${TIDB_VERSION}" cdc)
fi

until "${CDC_CMD[@]}" changefeed list --server="$TICDC_SERVER" >/dev/null 2>&1; do
  sleep 2
done
```

## 5. Seed TiDB And Dump Snapshot Directly To S3

```bash
mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root <<'SQL'
DROP DATABASE IF EXISTS test;
CREATE DATABASE test;
USE test;
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');
SQL

tiup dumpling \
  -u root \
  -P "$TIDB_PORT" \
  -h "$TIDB_HOST" \
  --filetype csv \
  -t 8 \
  -r 200000 \
  -F 64MiB \
  -B "$DB_NAME" \
  --s3.region "$S3_REGION" \
  -o "s3://$S3_BUCKET/$S3_RUN_PREFIX/snapshot"

```

Verify the snapshot objects are in S3:

```bash
"${AWS_S3[@]}" s3 ls "s3://$S3_BUCKET/$S3_RUN_PREFIX/snapshot/metadata"
"${AWS_S3[@]}" s3 ls "s3://$S3_BUCKET/$S3_RUN_PREFIX/snapshot/" --recursive | grep "$DB_NAME.$TABLE_NAME"
```

## 6. Deploy The Real Fivetran Connector

Create the Connector SDK configuration. This same S3 key pair is used by the
cloud connector to read the uploaded objects.

```bash
python3 - <<'PY'
from __future__ import annotations

import json
import os

s3_url = f"s3://{os.environ['S3_BUCKET']}/{os.environ['S3_RUN_PREFIX'].strip('/')}"

config = {
    "sync_task_id": os.environ["SYNC_TASK_ID"],
    "s3_url": s3_url,
    "s3_region": os.environ["S3_REGION"],
    "s3_ak": os.environ["S3_AK"],
    "s3_sk": os.environ["S3_SK"],
    "snapshot_prefix": "snapshot",
    "incremental_prefix": "increment",
    "checkpoint_interval_rows": "10",
    "enable_diagnostic_columns": "true",
    "incremental": json.dumps({"csv_output_old_value": True}, separators=(",", ":")),
}

os.makedirs(os.environ["RUNTIME_DIR"], exist_ok=True)
path = f"{os.environ['RUNTIME_DIR']}/fivetran_configuration.json"
with open(path, "w", encoding="utf-8") as handle:
    json.dump(config, handle, indent=2)
    handle.write("\n")
print(path)
PY

chmod 600 "$RUNTIME_DIR/fivetran_configuration.json"
```

Deploy:

```bash
fivetran deploy "$PWD" \
  --force \
  --api-key "$FIVETRAN_API_KEY_BASE64" \
  --destination "$FIVETRAN_DESTINATION_NAME" \
  --connection "$FIVETRAN_CONNECTION_NAME" \
  --configuration "$RUNTIME_DIR/fivetran_configuration.json" \
  --python-version 3.13
```

Resolve the Fivetran connection ID:

```bash
export FIVETRAN_CONNECTION_ID="$(
python3 - <<'PY'
from __future__ import annotations

import json
import os
from urllib import parse, request

base_url = os.environ.get("FIVETRAN_BASE_URL", "https://api.fivetran.com/v1").rstrip("/")
headers = {
    "Authorization": f"Basic {os.environ['FIVETRAN_API_KEY_BASE64']}",
    "Accept": "application/json;version=2",
}

def get(path: str) -> dict:
    req = request.Request(f"{base_url}{path}", headers=headers, method="GET")
    with request.urlopen(req, timeout=60) as resp:
        payload = json.load(resp)
    if payload.get("code") != "Success":
        raise SystemExit(payload)
    return payload

groups = get("/groups")["data"]["items"]
group_matches = [item for item in groups if item.get("name") == os.environ["FIVETRAN_DESTINATION_NAME"]]
if len(group_matches) != 1:
    raise SystemExit(f"destination match count is {len(group_matches)}")
group_id = group_matches[0]["id"]

query = parse.urlencode({"group_id": group_id})
connections = get(f"/connections?{query}")["data"]["items"]
connection_matches = [item for item in connections if item.get("schema") == os.environ["FIVETRAN_CONNECTION_NAME"]]
if len(connection_matches) != 1:
    raise SystemExit(f"connection match count is {len(connection_matches)}")
print(connection_matches[0]["id"])
PY
)"

echo "$FIVETRAN_CONNECTION_ID"
```

Trigger the snapshot sync:

```bash
python3 - <<'PY'
from __future__ import annotations

import json
import os
from urllib import request

base_url = os.environ.get("FIVETRAN_BASE_URL", "https://api.fivetran.com/v1").rstrip("/")
headers = {
    "Authorization": f"Basic {os.environ['FIVETRAN_API_KEY_BASE64']}",
    "Accept": "application/json;version=2",
    "Content-Type": "application/json",
}
body = json.dumps({"force": True}).encode("utf-8")
req = request.Request(
    f"{base_url}/connections/{os.environ['FIVETRAN_CONNECTION_ID']}/sync",
    data=body,
    headers=headers,
    method="POST",
)
with request.urlopen(req, timeout=60) as resp:
    print(resp.read().decode("utf-8"))
PY
```

Wait until Snowflake has 10 active snapshot rows in the fruit table. You can do
this in the Snowflake UI or with any Snowflake SQL client:

```sql
SELECT table_schema, table_name
FROM information_schema.tables
WHERE UPPER(table_schema) = UPPER('<SNOWFLAKE_SCHEMA>')
ORDER BY table_name;

SELECT COUNT(*)
FROM "<SNOWFLAKE_SCHEMA>"."TEST_FRUIT_1"
WHERE COALESCE(_FIVETRAN_DELETED, FALSE) = FALSE;
```

If the table name is `TEST_FRUIT1` instead of `TEST_FRUIT_1`, use that exact
table name in subsequent Snowflake queries.

## 7. Create The TiCDC Changefeed

TiCDC writes directly to S3. Do not put AK/SK in the sink URI; TiCDC reads the
AWS credentials from the environment inherited when `tiup playground` was
started.

```bash
cat > "$RUNTIME_DIR/changefeed.toml" <<'EOF'
[sink]
[sink.csv]
include-commit-ts = true
EOF

export TICDC_SINK_URI="s3://$S3_BUCKET/$S3_RUN_PREFIX/increment?protocol=csv&output-old-value=true"

"${CDC_CMD[@]}" changefeed create \
  --server="$TICDC_SERVER" \
  --sink-uri="$TICDC_SINK_URI" \
  --changefeed-id="$RUN_ID" \
  --config="$RUNTIME_DIR/changefeed.toml" \
  --no-confirm
```

## 8. Apply DDL Cases And Run `test.sql`

```bash
mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root <<'SQL'
USE test;
INSERT INTO fruit1(name) VALUES ('fruit-11');
ALTER TABLE fruit1 ADD COLUMN color varchar(20) DEFAULT NULL;
INSERT INTO fruit1(name, color) VALUES ('fruit-12', 'red');
ALTER TABLE fruit1 ADD COLUMN sweetness int NOT NULL DEFAULT 0;
INSERT INTO fruit1(name, color, sweetness) VALUES ('fruit-13', 'yellow', 7);
ALTER TABLE fruit1 MODIFY COLUMN name varchar(64) NOT NULL DEFAULT '';
INSERT INTO fruit1(name, color, sweetness) VALUES ('fruit-14-long-name-after-widen', 'purple', 8);
ALTER TABLE fruit1 COMMENT = 'snowflake e2e ddl coverage';
CREATE TABLE ddl_no_pk(note varchar(20));
INSERT INTO ddl_no_pk(note) VALUES ('ignored');
TRUNCATE TABLE fruit1;
INSERT INTO fruit1(id, name, color, sweetness) VALUES (100, 'fruit-after-truncate', 'green', 9);
SQL

mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root -e "SET GLOBAL tidb_enable_check_constraint = ON;" || true
mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root -e "SET GLOBAL tidb_enable_foreign_key = ON;" || true
mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root < e2e-test/test.sql
```

Wait until TiCDC has written incremental files to S3:

```bash
export INCREMENT_S3_URI="s3://$S3_BUCKET/$S3_RUN_PREFIX/increment"

until "${AWS_S3[@]}" s3 ls "$INCREMENT_S3_URI/metadata" >/dev/null 2>&1 &&
  "${AWS_S3[@]}" s3 ls "$INCREMENT_S3_URI/" --recursive | grep -E '/CDC[^/]*\.csv$' >/dev/null &&
  "${AWS_S3[@]}" s3 ls "$INCREMENT_S3_URI/" --recursive | grep -E '/schema_[^/]*\.json$' >/dev/null; do
  sleep 2
done
```

Optionally confirm the final marker rows are present in S3 CDC files:

```bash
for key in $("${AWS_S3[@]}" s3 ls "$INCREMENT_S3_URI/" --recursive | awk '/\/CDC[^/]*\.csv$/ {print $4}'); do
  "${AWS_S3[@]}" s3 cp "s3://$S3_BUCKET/$key" - --only-show-errors
done | grep -E 'fruit-after-truncate|cross-family value after ddl'
```

## 9. Trigger Fivetran Again

Trigger another Fivetran sync:

```bash
python3 - <<'PY'
from __future__ import annotations

import json
import os
from urllib import request

base_url = os.environ.get("FIVETRAN_BASE_URL", "https://api.fivetran.com/v1").rstrip("/")
headers = {
    "Authorization": f"Basic {os.environ['FIVETRAN_API_KEY_BASE64']}",
    "Accept": "application/json;version=2",
    "Content-Type": "application/json",
}
body = json.dumps({"force": True}).encode("utf-8")
req = request.Request(
    f"{base_url}/connections/{os.environ['FIVETRAN_CONNECTION_ID']}/sync",
    data=body,
    headers=headers,
    method="POST",
)
with request.urlopen(req, timeout=60) as resp:
    print(resp.read().decode("utf-8"))
PY
```

## 10. Compare TiDB And Snowflake

Export TiDB active-state profiles:

```bash
mkdir -p "$RUNTIME_DIR/comparison"

mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root --batch --raw --skip-column-names -e "
SELECT CAST(id AS CHAR), COALESCE(name, '__NULL__'), COALESCE(color, '__NULL__'), COALESCE(CAST(sweetness AS CHAR), '__NULL__')
FROM test.fruit1
ORDER BY id;
" > "$RUNTIME_DIR/comparison/fruit.tidb.tsv"

mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root --batch --raw --skip-column-names -e "
SELECT
  CAST(id AS CHAR),
  COALESCE(c_varchar, '__NULL__'),
  COALESCE(c_added_nullable, '__NULL__'),
  COALESCE(c_default_toggle, '__NULL__'),
  COALESCE(CAST(c_changed_rename_type AS CHAR), '__NULL__'),
  COALESCE(CAST(c_type_cross_family_number AS CHAR), '__NULL__')
FROM ft_ddl_type_test.ft_ddl_type_matrix
ORDER BY id;
" > "$RUNTIME_DIR/comparison/matrix.tidb.tsv"

mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root --batch --raw --skip-column-names -e "
SELECT CAST(id AS CHAR), COALESCE(payload, '__NULL__')
FROM ft_ddl_type_test.ft_ddl_create_after_snapshot
ORDER BY id;
" > "$RUNTIME_DIR/comparison/create.tidb.tsv"

mysql --protocol=tcp -h "$TIDB_HOST" -P "$TIDB_PORT" -u root --batch --raw --skip-column-names -e "
SELECT CAST(id AS CHAR), COALESCE(payload, '__NULL__')
FROM ft_ddl_type_test.ft_ddl_truncate_target
ORDER BY id;
" > "$RUNTIME_DIR/comparison/truncate.tidb.tsv"
```

In Snowflake, run the matching active-row queries and export results in the same
column order. Use the exact table names you found in `information_schema.tables`.

Fruit profile:

```sql
SELECT
  COALESCE(TO_VARCHAR(id), '__NULL__'),
  COALESCE(TO_VARCHAR(name), '__NULL__'),
  COALESCE(TO_VARCHAR(color), '__NULL__'),
  COALESCE(TO_VARCHAR(sweetness), '__NULL__')
FROM "<SNOWFLAKE_SCHEMA>"."TEST_FRUIT_1"
WHERE COALESCE(_FIVETRAN_DELETED, FALSE) = FALSE
ORDER BY TRY_TO_NUMBER(id), id;
```

DDL matrix profile:

```sql
SELECT
  COALESCE(TO_VARCHAR(id), '__NULL__'),
  COALESCE(TO_VARCHAR(c_varchar), '__NULL__'),
  COALESCE(TO_VARCHAR(c_added_nullable), '__NULL__'),
  COALESCE(TO_VARCHAR(c_default_toggle), '__NULL__'),
  COALESCE(TO_VARCHAR(c_changed_rename_type), '__NULL__'),
  COALESCE(TO_VARCHAR(c_type_cross_family_number), '__NULL__')
FROM "<SNOWFLAKE_SCHEMA>"."FT_DDL_TYPE_TEST_FT_DDL_TYPE_MATRIX"
WHERE COALESCE(_FIVETRAN_DELETED, FALSE) = FALSE
ORDER BY TRY_TO_NUMBER(id), id;
```

Create-after-snapshot profile:

```sql
SELECT
  COALESCE(TO_VARCHAR(id), '__NULL__'),
  COALESCE(TO_VARCHAR(payload), '__NULL__')
FROM "<SNOWFLAKE_SCHEMA>"."FT_DDL_TYPE_TEST_FT_DDL_CREATE_AFTER_SNAPSHOT"
WHERE COALESCE(_FIVETRAN_DELETED, FALSE) = FALSE
ORDER BY TRY_TO_NUMBER(id), id;
```

Truncate profile:

```sql
SELECT
  COALESCE(TO_VARCHAR(id), '__NULL__'),
  COALESCE(TO_VARCHAR(payload), '__NULL__')
FROM "<SNOWFLAKE_SCHEMA>"."FT_DDL_TYPE_TEST_FT_DDL_TRUNCATE_TARGET"
WHERE COALESCE(_FIVETRAN_DELETED, FALSE) = FALSE
ORDER BY TRY_TO_NUMBER(id), id;
```

Also verify no-primary-key tables are absent:

```sql
SELECT table_name
FROM information_schema.tables
WHERE UPPER(table_schema) = UPPER('<SNOWFLAKE_SCHEMA>')
  AND UPPER(table_name) IN (
    'TEST_DDL_NO_PK',
    'FT_DDL_TYPE_TEST_FT_DDL_NO_STABLE_PK'
  );
```

The query should return no rows.

## 11. Cleanup

Stop the local playground and remove TiUP data:

```bash
kill -TERM "$PLAYGROUND_PID" >/dev/null 2>&1 || true
sleep 2
kill -KILL "$PLAYGROUND_PID" >/dev/null 2>&1 || true
tiup clean "$TIUP_TAG" >/dev/null 2>&1 || true
```

Optional: remove this run's S3 objects after you finish debugging:

```bash
"${AWS_S3[@]}" s3 rm "s3://$S3_BUCKET/$S3_RUN_PREFIX" --recursive --only-show-errors
```

## Troubleshooting

- `===== end playground log tail =====` is only a separator printed after a
  failure. The real failure is usually above it, or in the first `ERROR:` line
  before the playground log is printed. Re-run with `PLAYGROUND_LOG_TAIL_LINES=400`
  if the useful part is truncated.
- `missing required command: fivetran` means the active Python environment does
  not have the Fivetran CLI. Run `UV_HTTP_TIMEOUT=120 uv sync`, then
  `source .venv/bin/activate`, then `command -v fivetran`.
- If TiDB never becomes ready, check for port conflicts on `4000`, `8300`,
  `2379`, or `3000`, then run `tiup clean "$TIUP_TAG"`.
- If Dumpling or TiCDC fails to write S3 objects, confirm
  `aws sts get-caller-identity` returns the intended E2E IAM identity, not your
  development identity. TiCDC reads `AWS_ACCESS_KEY_ID`,
  `AWS_SECRET_ACCESS_KEY`, and optional `AWS_SESSION_TOKEN` from the environment
  inherited when `tiup playground` starts.
- If Snowflake tables do not appear, confirm the Fivetran connection schema name
  equals `SNOWFLAKE_SCHEMA` and check the Fivetran dashboard for the latest sync
  error.