Repository: xataio/pgstream
Pull request #292 (2025-04-23 10:23:59)

Consolidate snapshot env vars
This PR consolidates the environment variables for the postgres listener and the snapshot listener. The same environment variables will be used for both, the decision of which mode will be used will be based on which url is provided (`PGSTREAM_POSTGRES_LISTENER_URL` or `PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL`). 

If both are provided, the stream configuration validation step will fail.


Patch:
diff --git a/cmd/config/config_env.go b/cmd/config/config_env.go
index 61866ac..06376b5 100644
--- a/cmd/config/config_env.go
+++ b/cmd/config/config_env.go
@@ -33,14 +33,6 @@ import (
 
 func init() {
 	viper.BindEnv("PGSTREAM_POSTGRES_LISTENER_URL")
-	viper.BindEnv("PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED")
-	viper.BindEnv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE")
-	viper.BindEnv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS")
-	viper.BindEnv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS")
-	viper.BindEnv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES")
-	viper.BindEnv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS")
-	viper.BindEnv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL")
-	viper.BindEnv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_REPEATABLE")
 	viper.BindEnv("PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME")
 
 	viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL")
@@ -104,6 +96,8 @@ func init() {
 
 	viper.BindEnv("PGSTREAM_INJECTOR_STORE_POSTGRES_URL")
 	viper.BindEnv("PGSTREAM_TRANSFORMER_RULES_FILE")
+	viper.BindEnv("PGSTREAM_FILTER_INCLUDE_TABLES")
+	viper.BindEnv("PGSTREAM_FILTER_EXCLUDE_TABLES")
 
 	viper.BindEnv("PGSTREAM_KAFKA_TLS_ENABLED")
 	viper.BindEnv("PGSTREAM_KAFKA_TLS_CA_CERT_FILE")
@@ -145,9 +139,9 @@ func parsePostgresListenerConfig() *stream.PostgresListenerConfig {
 		},
 	}
 
-	initialSnapshotEnabled := viper.GetBool("PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED")
-	if initialSnapshotEnabled {
-		cfg.Snapshot = parseSnapshotConfig(pgURL, "PGSTREAM_POSTGRES_INITIAL")
+	snapshotTables := viper.GetStringSlice("PGSTREAM_POSTGRES_SNAPSHOT_TABLES")
+	if len(snapshotTables) > 0 {
+		cfg.Snapshot = parseSnapshotConfig(pgURL)
 	}
 
 	return cfg
@@ -158,27 +152,27 @@ func parseSnapshotListenerConfig() *snapshotbuilder.SnapshotListenerConfig {
 	if pgsnapshotURL == "" {
 		return nil
 	}
-	return parseSnapshotConfig(pgsnapshotURL, "PGSTREAM_POSTGRES")
+	return parseSnapshotConfig(pgsnapshotURL)
 }
 
-func parseSnapshotConfig(pgURL, prefix string) *snapshotbuilder.SnapshotListenerConfig {
+func parseSnapshotConfig(pgURL string) *snapshotbuilder.SnapshotListenerConfig {
 	cfg := &snapshotbuilder.SnapshotListenerConfig{
 		Generator: pgsnapshotgenerator.Config{
 			URL:           pgURL,
-			BatchPageSize: viper.GetUint(fmt.Sprintf("%s_SNAPSHOT_BATCH_PAGE_SIZE", prefix)),
-			SchemaWorkers: viper.GetUint(fmt.Sprintf("%s_SNAPSHOT_SCHEMA_WORKERS", prefix)),
-			TableWorkers:  viper.GetUint(fmt.Sprintf("%s_SNAPSHOT_TABLE_WORKERS", prefix)),
+			BatchPageSize: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE"),
+			SchemaWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS"),
+			TableWorkers:  viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS"),
 		},
 		Adapter: adapter.SnapshotConfig{
-			Tables:          viper.GetStringSlice(fmt.Sprintf("%s_SNAPSHOT_TABLES", prefix)),
-			SnapshotWorkers: viper.GetUint(fmt.Sprintf("%s_SNAPSHOT_WORKERS", prefix)),
+			Tables:          viper.GetStringSlice("PGSTREAM_POSTGRES_SNAPSHOT_TABLES"),
+			SnapshotWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_WORKERS"),
 		},
-		Schema: parseSchemaSnapshotConfig(prefix, pgURL),
+		Schema: parseSchemaSnapshotConfig(pgURL),
 	}
 
-	if storeURL := viper.GetString(fmt.Sprintf("%s_SNAPSHOT_STORE_URL", prefix)); storeURL != "" {
+	if storeURL := viper.GetString("PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL"); storeURL != "" {
 		cfg.Recorder = &snapshotbuilder.SnapshotRecorderConfig{
-			RepeatableSnapshots: viper.GetBool(fmt.Sprintf("%s_SNAPSHOT_STORE_REPEATABLE", prefix)),
+			RepeatableSnapshots: viper.GetBool("PGSTREAM_POSTGRES_SNAPSHOT_STORE_REPEATABLE"),
 			SnapshotStoreURL:    storeURL,
 		}
 	}
@@ -186,8 +180,8 @@ func parseSnapshotConfig(pgURL, prefix string) *snapshotbuilder.SnapshotListener
 	return cfg
 }
 
-func parseSchemaSnapshotConfig(prefix, pgurl string) snapshotbuilder.SchemaSnapshotConfig {
-	useSchemaLog := viper.GetBool(fmt.Sprintf("%s_SNAPSHOT_USE_SCHEMALOG", prefix))
+func parseSchemaSnapshotConfig(pgurl string) snapshotbuilder.SchemaSnapshotConfig {
+	useSchemaLog := viper.GetBool("PGSTREAM_POSTGRES_SNAPSHOT_USE_SCHEMALOG")
 	pgTargetURL := viper.GetString("PGSTREAM_POSTGRES_WRITER_TARGET_URL")
 	if pgTargetURL != "" && !useSchemaLog {
 		return snapshotbuilder.SchemaSnapshotConfig{
diff --git a/docs/README.md b/docs/README.md
index 6bfbf15..11d287e 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -90,19 +90,18 @@ Here's a list of all the environment variables that can be used to configure the
 <details>
   <summary>Postgres Listener</summary>
 
-| Environment Variable                                | Default                | Required | Description                                                                                                                                                                                                                                                                                                  |
-| --------------------------------------------------- | ---------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
-| PGSTREAM_POSTGRES_LISTENER_URL                      | N/A                    | Yes      | URL of the Postgres database to connect to for replication purposes.                                                                                                                                                                                                                                         |
-| PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME             | "pgstream_dbname_slot" | No       | Name of the Postgres replication slot name.                                                                                                                                                                                                                                                                  |
-| PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED | false                  | No       | Enables performing an initial snapshot of the Postgres database before starting to listen to the replication slot.                                                                                                                                                                                           |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES           | ""                     | No       | Tables for which there will be an initial snapshot generated. The syntax supports wildcards. Tables without a schema defined will be applied the public schema. Example: for `public.test_table` and all tables in the `test_schema` schema, the value would be the following: `"test_table test_schema.\*"` |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS   | 4                      | No       | Number of tables per schema that will be processed in parallel by the snapshotting process.                                                                                                                                                                                                                  |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS    | 4                      | No       | Number of concurrent workers that will be used per table by the snapshotting process.                                                                                                                                                                                                                        |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE  | 1000                   | No       | Size of the table page range which will be processed concurrently by the table workers from `PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS`.                                                                                                                                                              |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS          | 1                      | No       | Number of schemas that will be processed in parallel by the snapshotting process.                                                                                                                                                                                                                            |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_USE_SCHEMALOG    | False                  | No       | Forces the use of the `pgstream.schema_log` for the schema snapshot instead of using `pg_dump`/`pg_restore` for Postgres outputs.                                                                                                                                                                            |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL        | ""                     | No       | Postgres URL for the database where the snapshot requests and their status will be tracked. A table `snapshot_requests` will be created under a `pgstream` schema.                                                                                                                                           |
-| PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_REPEATABLE | False                  | No       | Allow to repeat snapshots requests that have been already completed succesfully. By default snapshots that have been completed won't be repeated.                                                                                                                                                            |
+| Environment Variable                        | Default                | Required | Description                                                                                                                                                                                                                                                                                                  |
+| ------------------------------------------- | ---------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| PGSTREAM_POSTGRES_LISTENER_URL              | N/A                    | Yes      | URL of the Postgres database to connect to for replication purposes.                                                                                                                                                                                                                                         |
+| PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME     | "pgstream_dbname_slot" | No       | Name of the Postgres replication slot name.                                                                                                                                                                                                                                                                  |
+| PGSTREAM_POSTGRES_SNAPSHOT_TABLES           | ""                     | No       | Tables for which there will be an initial snapshot generated. The syntax supports wildcards. Tables without a schema defined will be applied the public schema. Example: for `public.test_table` and all tables in the `test_schema` schema, the value would be the following: `"test_table test_schema.\*"` |
+| PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS   | 4                      | No       | Number of tables per schema that will be processed in parallel by the snapshotting process.                                                                                                                                                                                                                  |
+| PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS    | 4                      | No       | Number of concurrent workers that will be used per table by the snapshotting process.                                                                                                                                                                                                                        |
+| PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE  | 1000                   | No       | Size of the table page range which will be processed concurrently by the table workers from `PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS`.                                                                                                                                                                      |
+| PGSTREAM_POSTGRES_SNAPSHOT_WORKERS          | 1                      | No       | Number of schemas that will be processed in parallel by the snapshotting process.                                                                                                                                                                                                                            |
+| PGSTREAM_POSTGRES_SNAPSHOT_USE_SCHEMALOG    | False                  | No       | Forces the use of the `pgstream.schema_log` for the schema snapshot instead of using `pg_dump`/`pg_restore` for Postgres outputs.                                                                                                                                                                            |
+| PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL        | ""                     | No       | Postgres URL for the database where the snapshot requests and their status will be tracked. A table `snapshot_requests` will be created under a `pgstream` schema.                                                                                                                                           |
+| PGSTREAM_POSTGRES_SNAPSHOT_STORE_REPEATABLE | False                  | No       | Allow to repeat snapshots requests that have been already completed succesfully. By default snapshots that have been completed won't be repeated.                                                                                                                                                            |
 
 </details>
 
diff --git a/docs/tutorials/postgres_kafka.md b/docs/tutorials/postgres_kafka.md
index 5768b5d..5082e5f 100644
--- a/docs/tutorials/postgres_kafka.md
+++ b/docs/tutorials/postgres_kafka.md
@@ -138,16 +138,14 @@ The PostgreSQL WAL listener can be configured to perform an initial snapshot of
 However, if there were tables with pre-existing data that we wanted to replicate to the target PostgreSQL database, we could configure it by setting the following environment variables:
 
 ```sh
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-
 # URL of the PostgreSQL database we want to snapshot
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 
 # List of tables we want to snapshot. If the tables are not schema qualified, the public schema will be assumed.
 # Wildcards are supported.
 #
 # The following example will snapshot all tables in the `test_schema` and the table `test` from the public schema.
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="test_schema.* test"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="test_schema.* test"
 ```
 
 Further configuration can be provided to optimize the performance of the snapshot process. For more information, check the [snapshot tutorial](postgres_snapshot).
@@ -220,10 +218,9 @@ PGSTREAM_KAFKA_TOPIC_AUTO_CREATE=true
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 # Initial snapshot of all tables in the public schema
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
diff --git a/docs/tutorials/postgres_to_opensearch.md b/docs/tutorials/postgres_to_opensearch.md
index a0ec2cb..98378ca 100644
--- a/docs/tutorials/postgres_to_opensearch.md
+++ b/docs/tutorials/postgres_to_opensearch.md
@@ -137,16 +137,14 @@ The PostgreSQL WAL listener can be configured to perform an initial snapshot of
 However, if there were tables with pre-existing data that we wanted to replicate to the target PostgreSQL database, we could configure it by setting the following environment variables:
 
 ```sh
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-
 # URL of the PostgreSQL database we want to snapshot
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 
 # List of tables we want to snapshot. If the tables are not schema qualified, the public schema will be assumed.
 # Wildcards are supported.
 #
 # The following example will snapshot all tables in the `test_schema` and the table `test` from the public schema.
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="test_schema.* test"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="test_schema.* test"
 ```
 
 Further configuration can be provided to optimize the performance of the snapshot process. For more information, check the [snapshot tutorial](postgres_snapshot).
@@ -214,10 +212,9 @@ PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT=5s
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 # Initial snapshot of all tables in the public schema
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
diff --git a/docs/tutorials/postgres_to_postgres.md b/docs/tutorials/postgres_to_postgres.md
index 0237449..d189897 100644
--- a/docs/tutorials/postgres_to_postgres.md
+++ b/docs/tutorials/postgres_to_postgres.md
@@ -125,14 +125,13 @@ PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
 If you want to perform an initial snapshot of existing tables, add:
 
 ```sh
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
 # URL of the PostgreSQL database we want to snapshot
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 # List of tables we want to snapshot. If the tables are not schema qualified, the public schema will be assumed.
 # Wildcards are supported.
 #
 # The following example will snapshot all tables in the `test_schema` and the table `test` from the public schema.
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="test_schema.* test"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="test_schema.* test"
 ```
 
 Further configuration can be provided to optimize the performance of the snapshot process. For more information, check the [snapshot tutorial](postgres_snapshot).
@@ -309,7 +308,7 @@ Here are some common issues you might encounter while following this tutorial an
 
 - **Cause:** The initial snapshot process encountered an issue.
 - **Solution:**
-  - Ensure the `PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED` variable is set to `true` if a snapshot is required.
+  - Ensure the `PGSTREAM_POSTGRES_SNAPSHOT_TABLES` variable is set if a snapshot is required.
   - Check the `pgstream` logs for detailed error messages:
     ```sh
     pgstream run -c pg2pg_tutorial.env --log-level trace
diff --git a/docs/tutorials/postgres_to_webhooks.md b/docs/tutorials/postgres_to_webhooks.md
index c12f385..fe8b064 100644
--- a/docs/tutorials/postgres_to_webhooks.md
+++ b/docs/tutorials/postgres_to_webhooks.md
@@ -138,16 +138,14 @@ The PostgreSQL WAL listener can be configured to perform an initial snapshot of
 However, if there were tables with pre-existing data that we wanted to replicate to the webhooks server, we could configure it by setting the following environment variables:
 
 ```sh
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-
 # URL of the PostgreSQL database we want to snapshot
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 
 # List of tables we want to snapshot. If the tables are not schema qualified, the public schema will be assumed.
 # Wildcards are supported.
 #
 # The following example will snapshot all tables in the `test_schema` and the table `test` from the public schema.
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="test_schema.* test"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="test_schema.* test"
 ```
 
 Further configuration can be provided to optimize the performance of the snapshot process. For more information, check the [snapshot tutorial](postgres_snapshot#listener).
@@ -189,10 +187,9 @@ PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_CACHE_REFRESH_INTERVAL="60s"
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 # Initial snapshot of all tables in the public schema
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
 
 # Processor config
 PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
diff --git a/docs/tutorials/postgres_transformer.md b/docs/tutorials/postgres_transformer.md
index 27fe5f5..73a1c37 100644
--- a/docs/tutorials/postgres_transformer.md
+++ b/docs/tutorials/postgres_transformer.md
@@ -136,10 +136,9 @@ PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 # Initial snapshot of all tables in the public schema
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
 ```
 
 ### Processor
@@ -202,10 +201,9 @@ PGSTREAM_POSTGRES_WRITER_SCHEMALOG_STORE_URL="postgres://postgres:postgres@local
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
 # Initial snapshot of all tables in the public schema
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
 
 # Processor config
 PGSTREAM_TRANSFORMER_RULES_FILE="tutorial_transformer_rules.yaml"
diff --git a/pg2kafka.env b/pg2kafka.env
index 98db563..abe5d2a 100644
--- a/pg2kafka.env
+++ b/pg2kafka.env
@@ -1,11 +1,10 @@
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost?sslmode=disable"
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE=1000
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS=1
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE=1000
+PGSTREAM_POSTGRES_SNAPSHOT_WORKERS=1
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost?sslmode=disable"
diff --git a/pg2os.env b/pg2os.env
index dc657dd..f53014f 100644
--- a/pg2os.env
+++ b/pg2os.env
@@ -1,11 +1,10 @@
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost?sslmode=disable"
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE=1000
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS=1
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE=1000
+PGSTREAM_POSTGRES_SNAPSHOT_WORKERS=1
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost?sslmode=disable"
diff --git a/pg2pg.env b/pg2pg.env
index 0889d50..a011c59 100644
--- a/pg2pg.env
+++ b/pg2pg.env
@@ -1,12 +1,11 @@
 # Listener config
 PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost?sslmode=disable"
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost?sslmode=disable"
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="*"
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE=1000
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS=1
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgres://postgres:postgres@localhost?sslmode=disable"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE=1000
+PGSTREAM_POSTGRES_SNAPSHOT_WORKERS=1
 
 # Processor config
 PGSTREAM_TRANSFORMER_RULES_FILE="transformer_rules.yaml"


Test patch:
diff --git a/cmd/config/config_env_test.go b/cmd/config/config_env_test.go
index b787543..b61e79c 100644
--- a/cmd/config/config_env_test.go
+++ b/cmd/config/config_env_test.go
@@ -23,14 +23,13 @@ func Test_EnvConfigToStreamConfig(t *testing.T) {
 func Test_EnvVarsToStreamConfig(t *testing.T) {
 	os.Setenv("PGSTREAM_POSTGRES_LISTENER_URL", "postgresql://user:password@localhost:5432/mydatabase")
 	os.Setenv("PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME", "pgstream_mydatabase_slot")
-	os.Setenv("PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED", "true")
-	os.Setenv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL", "postgresql://user:password@localhost:5432/mytargetdatabase")
-	os.Setenv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_REPEATABLE", "true")
-	os.Setenv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES", "test test_schema.test another_schema.*")
-	os.Setenv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS", "4")
-	os.Setenv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS", "4")
-	os.Setenv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS", "4")
-	os.Setenv("PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE", "1000")
+	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL", "postgresql://user:password@localhost:5432/mytargetdatabase")
+	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_STORE_REPEATABLE", "true")
+	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_TABLES", "test test_schema.test another_schema.*")
+	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_WORKERS", "4")
+	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS", "4")
+	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS", "4")
+	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE", "1000")
 	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_CLEAN_TARGET_DB", "true")
 
 	os.Setenv("PGSTREAM_KAFKA_SERVERS", "localhost:9092")
@@ -82,6 +81,8 @@ func Test_EnvVarsToStreamConfig(t *testing.T) {
 
 	os.Setenv("PGSTREAM_INJECTOR_STORE_POSTGRES_URL", "postgresql://user:password@localhost:5432/mydatabase")
 	os.Setenv("PGSTREAM_TRANSFORMER_RULES_FILE", "test/test_transformer_rules.yaml")
+	os.Setenv("PGSTREAM_FILTER_INCLUDE_TABLES", "test test_schema.test another_schema.*")
+	os.Setenv("PGSTREAM_FILTER_EXCLUDE_TABLES", "excluded_test excluded_schema.test another_excluded_schema.*")
 
 	streamConfig, err := envConfigToStreamConfig()
 	assert.NoError(t, err)
diff --git a/cmd/config/test/test_config.env b/cmd/config/test/test_config.env
index c7ce928..6257ef4 100644
--- a/cmd/config/test/test_config.env
+++ b/cmd/config/test/test_config.env
@@ -3,14 +3,13 @@
 # Postgres
 PGSTREAM_POSTGRES_LISTENER_URL="postgresql://user:password@localhost:5432/mydatabase"
 PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME="pgstream_mydatabase_slot"
-PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgresql://user:password@localhost:5432/mytargetdatabase"
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_REPEATABLE=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="test test_schema.test another_schema.*"
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS=4
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE=1000
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL="postgresql://user:password@localhost:5432/mytargetdatabase"
+PGSTREAM_POSTGRES_SNAPSHOT_STORE_REPEATABLE=true
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="test test_schema.test another_schema.*"
+PGSTREAM_POSTGRES_SNAPSHOT_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS=4
+PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE=1000
 PGSTREAM_POSTGRES_SNAPSHOT_CLEAN_TARGET_DB=true
 
 # Kafka