Trendyol · Abdulsametileri · Nov 24, 2025 · Oct 16, 2025 · Oct 16, 2025 · Oct 18, 2025
diff --git a/.golangci.yml b/.golangci.yml
@@ -1,6 +1,6 @@
 linters-settings:
   funlen:
-    lines: 70
+    lines: 85
 
 linters:
   disable-all: true

diff --git a/README.md b/README.md
@@ -7,17 +7,35 @@ ensuring low resource consumption and high performance.
 
 [Debezium vs go-pq-cdc benchmark](./benchmark)
 
+## 📸 NEW: Snapshot Feature
+
+**Capture existing data before starting CDC!** The new snapshot feature enables initial data synchronization, ensuring downstream systems receive both historical and real-time data.
+
+✨ **Key Highlights:**
+- **Zero Data Loss**: Consistent point-in-time snapshot using PostgreSQL's `pg_export_snapshot()`
+- **Chunk-Based Processing**: Memory-efficient processing of large tables
+- **Multi-Instance Support**: Parallel processing across multiple instances
+- **Crash Recovery**: Automatic resume from failures
+- **No Duplicates**: Seamless transition from snapshot to CDC
+- **Snapshot Only Mode**: One-time data export without CDC (no replication slot required)
+
+📚 **[Read Full Documentation](docs/SNAPSHOT_FEATURE.md)** for detailed architecture, configuration, and best practices.
+
 ### Contents
 
-* [Why?](#why)
-* [Usage](#usage)
-* [Examples](#examples)
-* [Availability](#availability)
-* [Configuration](#configuration)
-* [API](#api)
-* [Exposed Metrics](#exposed-metrics)
-* [Compatibility](#compatibility)
-* [Breaking Changes](#breaking-changes)
+- [go-pq-cdc   ](#go-pq-cdc---)
+	- [📸 NEW: Snapshot Feature](#-new-snapshot-feature)
+		- [Contents](#contents)
+		- [Why?](#why)
+		- [Usage](#usage)
+		- [Examples](#examples)
+		- [Availability](#availability)
+		- [Configuration](#configuration)
+		- [API](#api)
+		- [Exposed Metrics](#exposed-metrics)
+		- [Grafana Dashboard](#grafana-dashboard)
+		- [Compatibility](#compatibility)
+		- [Breaking Changes](#breaking-changes)
 
 ### Why?
 
@@ -114,6 +132,8 @@ func Handler(ctx *replication.ListenerContext) {
 
 * [Simple](./example/simple)
 * [Simple File Config](./example/simple-file-config)
+* [Snapshot Mode (Initial Data Capture)](./example/snapshotmode)
+* [Snapshot Only Mode (One-Time Export)](./example/snapshotonlymode)
 * [PostgreSQL to Elasticsearch](https://github.com/Trendyol/go-pq-cdc-elasticsearch/tree/main/example/simple)
 * [PostgreSQL to Kafka](https://github.com/Trendyol/go-pq-cdc-kafka/tree/main/example/simple)
 * [PostgreSQL to PostgreSQL](./example/postgresql)
@@ -154,6 +174,13 @@ This setup ensures continuous data synchronization and minimal downtime in captu
 | `slot.createIfNotExists`                |   bool   |    no    |    -    | Create replication slot if not exists. Otherwise, return `replication slot is not exists` error.      |                                                                                                                                                    |
 | `slot.name`                             |  string  |   yes    |    -    | Set the logical replication slot name                                                                 | Should be unique and descriptive.                                                                                                                  |
 | `slot.slotActivityCheckerInterval`      |   int    |    no    |  1000   | Set the slot activity check interval time in milliseconds                                             | Specify as an integer value in milliseconds (e.g., `1000` for 1 second).                                                                           |
+| `snapshot.enabled`                      |   bool   |    no    |  false  | Enable initial snapshot feature                                                                       | When enabled, captures existing data before starting CDC.                                                                                          |
+| `snapshot.mode`                         |  string  |    no    |  never  | Snapshot mode: `initial`, `never`, or `snapshot_only`                                                 | **initial:** Take snapshot only if no previous snapshot exists, then start CDC. <br> **never:** Skip snapshot, start CDC immediately. <br> **snapshot_only:** Take snapshot and exit (no CDC, no replication slot required). |
+| `snapshot.chunkSize`                    |  int64   |    no    |  8000   | Number of rows per chunk during snapshot                                                              | Adjust based on table size. Larger chunks = fewer chunks but more memory per chunk.                                                               |
+| `snapshot.claimTimeout`                 | duration |    no    |  30s    | Timeout to reclaim stale chunks                                                                       | If a worker doesn't send heartbeat for this duration, chunk is reclaimed by another worker.                                                        |
+| `snapshot.heartbeatInterval`            | duration |    no    |  5s     | Interval for worker heartbeat updates                                                                 | Workers send heartbeat every N seconds to indicate they're processing a chunk.                                                                     |
+| `snapshot.instanceId`                   |  string  |    no    |  auto   | Custom instance identifier (optional)                                                                 | Auto-generated as `hostname-pid` if not specified. Useful for tracking workers.                                                                    |
+| `snapshot.tables`                       | []Table  |    no*   |    -    | Tables to snapshot (required for `snapshot_only` mode, optional for `initial` mode)                  | **snapshot_only:** Must be specified here (independent from publication). <br> **initial:** If specified, must be a subset of publication tables. If not specified, all publication tables are snapshotted. |
 | `extensionSupport.enableTimescaleDB`    |   bool   |    no    |  false  | Enable support for TimescaleDB hypertables. Ensures proper handling of compressed chunks during replication. |                                                                                                                                                    |
 
 ### API
@@ -181,6 +208,12 @@ the `/metrics` endpoint.
 | go_pq_cdc_replication_slot_slot_is_active           | Indicates whether the PostgreSQL replication slot is currently active (1 for active, 0 for inactive). | slot_name, host| Gauge      |
 | go_pq_cdc_replication_slot_slot_lag                 | The replication lag measured by the difference between the current LSN and the confirmed flush LSN.   | slot_name, host| Gauge      |
 | go_pq_cdc_replication_slot_slot_retained_wal_size   | The size of Write-Ahead Logging (WAL) files retained for the replication slot in bytes.               | slot_name, host| Gauge      |
+| go_pq_cdc_snapshot_in_progress                      | Indicates whether snapshot is currently in progress (1 for active, 0 for inactive).                   | slot_name, host| Gauge      |
+| go_pq_cdc_snapshot_total_tables                     | Total number of tables to snapshot.                                                                    | slot_name, host| Gauge      |
+| go_pq_cdc_snapshot_total_chunks                     | Total number of chunks to process across all tables.                                                   | slot_name, host| Gauge      |
+| go_pq_cdc_snapshot_completed_chunks                 | Number of chunks completed in snapshot.                                                                | slot_name, host| Gauge      |
+| go_pq_cdc_snapshot_total_rows                       | Total number of rows read during snapshot.                                                             | slot_name, host| Counter    |
+| go_pq_cdc_snapshot_duration_seconds                 | Duration of the last snapshot operation in seconds.                                                    | slot_name, host| Gauge      |
 | runtime metrics                                     | [Prometheus Collector](https://golang.bg/src/runtime/metrics/description.go)                          | N/A            | N/A        |
 
 ### Grafana Dashboard

diff --git a/benchmark/10m_test.png → benchmark/benchmark_cdc/10m_test.png b/benchmark/10m_test.png → benchmark/benchmark_cdc/10m_test.png
diff --git a/benchmark/README.md → benchmark/benchmark_cdc/README.md b/benchmark/README.md → benchmark/benchmark_cdc/README.md
diff --git a/benchmark/dashboard.png → benchmark/benchmark_cdc/dashboard.png b/benchmark/dashboard.png → benchmark/benchmark_cdc/dashboard.png
diff --git a/benchmark/docker-compose.yml → benchmark/benchmark_cdc/docker-compose.yml b/benchmark/docker-compose.yml → benchmark/benchmark_cdc/docker-compose.yml
diff --git a/benchmark/go-pq-cdc → benchmark/benchmark_cdc/go-pq-cdc b/benchmark/go-pq-cdc → benchmark/benchmark_cdc/go-pq-cdc
diff --git a/benchmark/go-pq-cdc-kafka/Dockerfile → .../benchmark_cdc/go-pq-cdc-kafka/Dockerfile b/benchmark/go-pq-cdc-kafka/Dockerfile → .../benchmark_cdc/go-pq-cdc-kafka/Dockerfile
diff --git a/benchmark/go-pq-cdc-kafka/go-pq-cdc-kafka → ...hmark_cdc/go-pq-cdc-kafka/go-pq-cdc-kafka b/benchmark/go-pq-cdc-kafka/go-pq-cdc-kafka → ...hmark_cdc/go-pq-cdc-kafka/go-pq-cdc-kafka
diff --git a/benchmark/go-pq-cdc-kafka/go.mod → ...mark/benchmark_cdc/go-pq-cdc-kafka/go.mod b/benchmark/go-pq-cdc-kafka/go.mod → ...mark/benchmark_cdc/go-pq-cdc-kafka/go.mod
@@ -2,7 +2,7 @@ module github.com/Trendyol/go-pq-cdc/benchmark/go-pq-cdc-kafka
 
 go 1.22.4
 
-replace github.com/Trendyol/go-pq-cdc => ../../
+replace github.com/Trendyol/go-pq-cdc => ../../../
 
 require (
 	github.com/Trendyol/go-pq-cdc v0.0.5

diff --git a/benchmark/go-pq-cdc-kafka/go.sum → ...mark/benchmark_cdc/go-pq-cdc-kafka/go.sum b/benchmark/go-pq-cdc-kafka/go.sum → ...mark/benchmark_cdc/go-pq-cdc-kafka/go.sum
diff --git a/benchmark/go-pq-cdc-kafka/main.go → ...ark/benchmark_cdc/go-pq-cdc-kafka/main.go b/benchmark/go-pq-cdc-kafka/main.go → ...ark/benchmark_cdc/go-pq-cdc-kafka/main.go
@@ -84,6 +84,7 @@ func main() {
 		os.Exit(1)
 	}
 
+	defer connector.Close()
 	connector.Start(ctx)
 }
 

diff --git a/benchmark/grafana/dashboard.yml → ...hmark/benchmark_cdc/grafana/dashboard.yml b/benchmark/grafana/dashboard.yml → ...hmark/benchmark_cdc/grafana/dashboard.yml
diff --git a/benchmark/grafana/dashboards/benchmark.json → ...ark_cdc/grafana/dashboards/benchmark.json b/benchmark/grafana/dashboards/benchmark.json → ...ark_cdc/grafana/dashboards/benchmark.json
diff --git a/benchmark/grafana/dashboards/dashboard.json → ...ark_cdc/grafana/dashboards/dashboard.json b/benchmark/grafana/dashboards/dashboard.json → ...ark_cdc/grafana/dashboards/dashboard.json
diff --git a/...ark/grafana/dashboards/pg_monitoring.json → ...cdc/grafana/dashboards/pg_monitoring.json b/...ark/grafana/dashboards/pg_monitoring.json → ...cdc/grafana/dashboards/pg_monitoring.json
diff --git a/benchmark/grafana/datasource.yml → ...mark/benchmark_cdc/grafana/datasource.yml b/benchmark/grafana/datasource.yml → ...mark/benchmark_cdc/grafana/datasource.yml
diff --git a/benchmark/grafana/grafana.ini → benchmark/benchmark_cdc/grafana/grafana.ini b/benchmark/grafana/grafana.ini → benchmark/benchmark_cdc/grafana/grafana.ini
diff --git a/benchmark/prometheus.yml → benchmark/benchmark_cdc/prometheus.yml b/benchmark/prometheus.yml → benchmark/benchmark_cdc/prometheus.yml
diff --git a/benchmark/sql/init.sql → benchmark/benchmark_cdc/sql/init.sql b/benchmark/sql/init.sql → benchmark/benchmark_cdc/sql/init.sql
diff --git a/benchmark/benchmark_initial/10m_test.png b/benchmark/benchmark_initial/10m_test.png
diff --git a/benchmark/benchmark_initial/README.md b/benchmark/benchmark_initial/README.md
@@ -0,0 +1,78 @@
+## 10 M Insert Test
+
+### Hardware 
+```txt
+PC: Macbook Apple M1 Pro (2021)
+Memory: 32 GB
+
+go-pq-cdc: 
+  resources:
+    limits:
+      cpus: 1
+      memory: 512M
+    reservations:
+      cpus: '0.25'
+      memory: 128M
+
+Debezium:
+  resources:
+    limits:
+      cpus: 2
+      memory: 1024M
+    reservations:
+      cpus: '0.25'
+      memory: 128M
+```
+
+### Result
+|                      | go-pq-cdc  | Debezium     |
+|----------------------|------------|--------------|
+| Row Count            | 10 m       | 10 m         |
+| Elapsed Time         | 2.5 min    | 21 min       |
+| Cpu Usage Max        | 44%        | 181%         |
+| Memory Usage Max     | 130 MB     | 1.07 GB      |
+| Received Traffic Max | 4.36 MiB/s | 7.97 MiB/s   |
+| Sent Traffic Max     | 5.96 MiB/s | 6.27 MiB/s   |
+
+![10m_result](./10m_test.png)
+
+
+## Requirements
+- [Docker](https://docs.docker.com/compose/install/)
+- [psql](https://www.postgresql.org/download/)
+
+## Instructions
+
+- Start the containers 
+    ```sh
+    docker compose up -d
+    ```
+- Connect to Postgres database:
+   ```sh
+   psql postgres://cdc_user:[email protected]:5432/cdc_db
+   ```
+- Insert data to users table: 
+    ```sql
+    INSERT INTO users (name)
+    SELECT
+        'Oyleli' || i
+    FROM generate_series(1, 1000000) AS i;
+    ```
+- Go to grafana dashboard: http://localhost:3000/d/edl1ybvsmc64gb/benchmark?orgId=1
+    > **Grafana Credentials**  
+     Username: `go-pq-cdc-user` Password: `go-pq-cdc-pass`
+- Trace the process 
+![benchmark_dashboard](./dashboard.png)
+
+## Ports
+
+- RedPanda Console: `8085` http://localhost:8085 
+- RedPanda: `19092` http://localhost:19092
+- Grafana: `3000` http://localhost:3000
+- Prometheus: `9090`  http://localhost:9090
+- cAdvisor: `8080` http://localhost:8080
+- PostgreSQL:`5432` http://localhost:5432
+- PostgreSQL Metric Exporter: `9187` http://localhost:9187 
+- Debezium: `9093` http://localhost:9093
+- go-pq-cdc Metric: `2112` http://localhost:2112
+
diff --git a/benchmark/benchmark_initial/dashboard.png b/benchmark/benchmark_initial/dashboard.png
-Original file line number
+Diff line change
@@ Expand Up / @@ -84,6 +84,7 @@ func main() { @@
     		os.Exit(1)
     	}
+    	defer connector.Close()
     	connector.Start(ctx)
     }
@@ Expand Down @@