Repository: xataio/pgstream
Pull request #275 (2025-04-14 15:54:34)

Add table filtering support
This PR adds support to configure the pgstream filtering using a list to either include or exclude tables. By default the `pgstream.schema_log` table is in the include list, since it's required to track and replicate DDL changes. 

The filtering logic has been removed from the injector, and it's now part of the new filter processor wrapper.

github-actions[bot]: ### Merging this branch will **increase** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd/config | 80.72% (**+0.22%**) | :thumbsup: |
| github.com/xataio/pgstream/pkg/stream | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/wal/processor/filter | 94.20% (**+94.20%**) | :star2: |
| github.com/xataio/pgstream/pkg/wal/processor/injector | 73.96% (**+3.20%**) | :thumbsup: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/config/config_env.go | 81.40% (**-0.09%**) | 86 (+5) | 70 (+4) | 16 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/cmd/config/config_yaml.go | 89.15% (**+0.26%**) | 129 (+3) | 115 (+3) | 14 | :thumbsup: |
| github.com/xataio/pgstream/pkg/stream/config.go | 0.00% (ø) | 27 | 0 | 27 |  |
| github.com/xataio/pgstream/pkg/stream/stream_run.go | 0.00% (ø) | 166 (+6) | 0 | 166 (+6) |  |
| github.com/xataio/pgstream/pkg/wal/processor/filter/wal_filter.go | 94.20% (**+94.20%**) | 69 (+69) | 65 (+65) | 4 (+4) | :star2: |
| github.com/xataio/pgstream/pkg/wal/processor/injector/wal_injector.go | 73.96% (**+3.20%**) | 96 (-10) | 71 (-4) | 25 (-6) | :thumbsup: |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/cmd/config/helper_test.go
- github.com/xataio/pgstream/pkg/wal/processor/filter/wal_filter_test.go
- github.com/xataio/pgstream/pkg/wal/processor/injector/wal_injector_test.go

</details>


Patch:
diff --git a/README.md b/README.md
index 9b9d90c..d424211 100644
--- a/README.md
+++ b/README.md
@@ -171,7 +171,7 @@ Some of the limitations of the initial release include:
 
 - Single Kafka topic support
 - Postgres plugin support limited to `wal2json`
-- Limited filtering
+- No row level filtering support
 - Primary key/unique not null column required for replication
 - Kafka serialisation support limited to JSON
 
diff --git a/cmd/config/config_env.go b/cmd/config/config_env.go
index c782f82..61866ac 100644
--- a/cmd/config/config_env.go
+++ b/cmd/config/config_env.go
@@ -18,6 +18,7 @@ import (
 	"github.com/xataio/pgstream/pkg/wal/listener/snapshot/adapter"
 	snapshotbuilder "github.com/xataio/pgstream/pkg/wal/listener/snapshot/builder"
 	"github.com/xataio/pgstream/pkg/wal/processor/batch"
+	"github.com/xataio/pgstream/pkg/wal/processor/filter"
 	"github.com/xataio/pgstream/pkg/wal/processor/injector"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
 	"github.com/xataio/pgstream/pkg/wal/processor/postgres"
@@ -252,6 +253,7 @@ func parseProcessorConfig() (stream.ProcessorConfig, error) {
 		Postgres:    parsePostgresProcessorConfig(),
 		Injector:    parseInjectorConfig(),
 		Transformer: transformerCfg,
+		Filter:      parseFilterConfig(),
 	}, nil
 }
 
@@ -431,6 +433,19 @@ func parseTransformerConfig() (*transformer.Config, error) {
 	return yamlConfig.Transformations.parseTransformationConfig(), nil
 }
 
+func parseFilterConfig() *filter.Config {
+	includeTables := viper.GetStringSlice("PGSTREAM_FILTER_INCLUDE_TABLES")
+	excludeTables := viper.GetStringSlice("PGSTREAM_FILTER_EXCLUDE_TABLES")
+	if len(includeTables) == 0 && len(excludeTables) == 0 {
+		return nil
+	}
+
+	return &filter.Config{
+		IncludeTables: includeTables,
+		ExcludeTables: excludeTables,
+	}
+}
+
 func parseTLSConfig(prefix string) tls.Config {
 	return tls.Config{
 		Enabled:        viper.GetBool(fmt.Sprintf("%s_TLS_ENABLED", prefix)),
diff --git a/cmd/config/config_yaml.go b/cmd/config/config_yaml.go
index 0a74deb..0e50580 100644
--- a/cmd/config/config_yaml.go
+++ b/cmd/config/config_yaml.go
@@ -18,6 +18,7 @@ import (
 	"github.com/xataio/pgstream/pkg/wal/listener/snapshot/adapter"
 	snapshotbuilder "github.com/xataio/pgstream/pkg/wal/listener/snapshot/builder"
 	"github.com/xataio/pgstream/pkg/wal/processor/batch"
+	"github.com/xataio/pgstream/pkg/wal/processor/filter"
 	"github.com/xataio/pgstream/pkg/wal/processor/injector"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
 	"github.com/xataio/pgstream/pkg/wal/processor/postgres"
@@ -198,6 +199,7 @@ type WebhookNotifierConfig struct {
 type ModifiersConfig struct {
 	Injector        *InjectorConfig       `mapstructure:"injector" yaml:"injector"`
 	Transformations TransformationsConfig `mapstructure:"transformations" yaml:"transformations"`
+	Filter          *FilterConfig         `mapstructure:"filter" yaml:"filter"`
 }
 
 type InjectorConfig struct {
@@ -205,6 +207,11 @@ type InjectorConfig struct {
 	SchemalogURL string `mapstructure:"schemalog_url" yaml:"schemalog_url"`
 }
 
+type FilterConfig struct {
+	IncludeTables []string `mapstructure:"include_tables" yaml:"include_tables"`
+	ExcludeTables []string `mapstructure:"exclude_tables" yaml:"exclude_tables"`
+}
+
 type TransformationsConfig []TransformationConfig
 
 type TransformationConfig struct {
@@ -289,6 +296,7 @@ func (c *YAMLConfig) parseProcessorConfig() (stream.ProcessorConfig, error) {
 		Postgres:    c.parsePostgresProcessorConfig(),
 		Webhook:     c.parseWebhookProcessorConfig(),
 		Transformer: c.parseTransformationConfig(),
+		Filter:      c.parseFilterConfig(),
 	}
 
 	var err error
@@ -546,6 +554,16 @@ func (c *YAMLConfig) parseTransformationConfig() *transformer.Config {
 	return c.Modifiers.Transformations.parseTransformationConfig()
 }
 
+func (c YAMLConfig) parseFilterConfig() *filter.Config {
+	if c.Modifiers.Filter == nil {
+		return nil
+	}
+	return &filter.Config{
+		ExcludeTables: c.Modifiers.Filter.ExcludeTables,
+		IncludeTables: c.Modifiers.Filter.IncludeTables,
+	}
+}
+
 func (c TransformationsConfig) parseTransformationConfig() *transformer.Config {
 	if len(c) == 0 {
 		return nil
diff --git a/config_template.yaml b/config_template.yaml
index fde5a95..9f3e373 100644
--- a/config_template.yaml
+++ b/config_template.yaml
@@ -101,6 +101,11 @@ modifiers:
   injector:
     enabled: true # whether to inject pgstream metadata into the WAL events
     schemalog_url: "postgresql://user:password@localhost:5432/mydatabase" # URL of the schemalog database, if different from the source database
+  filter: # one of include_tables or exclude_tables
+    include_tables: # list of tables for which events should be allowed. Tables should be schema qualified. If no schema is provided, the public schema will be assumed. Wildcards "*" are supported.
+      - "test"
+    exclude_tables: # list of tables for which events should be skipped. Tables should be schema qualified. If no schema is provided, the public schema will be assumed. Wildcards "*" are supported.
+      - "another_test"
   transformations:
   - schema: public
     table: test
diff --git a/docs/README.md b/docs/README.md
index c1788d7..6bfbf15 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -60,9 +60,11 @@ The current implementations of the processor include:
 
 - **Postgres batch writer**: it writes the WAL events into a PostgreSQL compatible database. It implements the same kind of mechanism than the Kafka and the search batch writers to ensure continuous processing from the listener, and it also uses a batching mechanism to minimise PostgreSQL IO traffic.
 
-In addition to the implementations described above, there are optional processor decorators, which work in conjunction with one of the main processor implementations described above. Their goal is to act as modifiers to enrich the wal event being processed.
+In addition to the implementations described above, there are optional processor decorators, which work in conjunction with one of the main processor implementations described above. Their goal is to act as modifiers to enrich the wal event being processed. We will refer to them as modifiers.
 
-There are currently two implementations of the processor that act as decorators:
+#### Modifiers
+
+There current implementations of the processor that act as modifier decorators are:
 
 - **Injector**: injects some of the pgstream logic into the WAL event. This includes:
 
@@ -75,6 +77,8 @@ There are currently two implementations of the processor that act as decorators:
   - Schema events:
     - Acknolwedging the new incoming schema in the Postgres `pgstream.schema_log` table.
 
+- **Filter**: allows to filter out WAL events for certain schemas/tables. It can be configured by providing either an include or exclude table list. WAL events for the tables in the include list will be processed, while those for the tables in the exclude list or not present in the include list will be skipped. The format for the lists is similar to the snapshot tables, tables are expected to be schema qualified, and if not, the `public` schema will be assumed. Wildcards are supported, but not regex. Example of table list: `["test_table", "public.test_table", "test_schema.test_table", "test_schema.*", "*.test", "*.*"]`. By default, the filter will include the `pgstream.schema_log` table in the include table list, since pgstream relies on it to replicate DDL changes. It can be disabled by adding it to the exclude list.
+
 - **Transformer**: it modifies the column values in insert/update events according to the rules defined in the configured yaml file. It can be used for anonymising data from the source Postgres database. An example of the rules definition file can be found in the repo under `transformer_rules.yaml`. The rules have per column granularity, and certain transformers from opensource sources, such as greenmask or neosync, are supported. More details can be found in the [transformers section](#transformers).
 
 ## Configuration
@@ -219,6 +223,8 @@ One of exponential/constant backoff policies can be provided for the search stor
 
 </details>
 
+#### Modifiers
+
 <details>
   <summary>Injector</summary>
 
@@ -237,6 +243,16 @@ One of exponential/constant backoff policies can be provided for the search stor
 
 </details>
 
+<details>
+  <summary>Filter</summary>
+
+| Environment Variable           | Default | Required | Description                                                                                                                                                       |
+| ------------------------------ | ------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| PGSTREAM_FILTER_INCLUDE_TABLES | N/A     | No       | List of schema qualified tables for which the WAL events should be processed. If no schema is provided, `public` schema will be assumed. Wildcards are supported. |
+| PGSTREAM_FILTER_EXCLUDE_TABLES | N/A     | No       | List of schema qualified tables for which the WAL events should be skipped. If no schema is provided, `public` schema will be assumed. Wildcards are supported.   |
+
+</details>
+
 ## Tracking schema changes
 
 One of the main differentiators of pgstream is the fact that it tracks and replicates schema changes automatically. It relies on SQL triggers that will populate a Postgres table (`pgstream.schema_log`) containing a history log of all DDL changes for a given schema. Whenever a schema change occurs, this trigger creates a new row in the schema log table with the schema encoded as a JSON value. This table tracks all the schema changes, forming a linearised change log that is then parsed and used within the pgstream pipeline to identify modifications and push the relevant changes downstream.
@@ -312,6 +328,7 @@ transformations:
 | `Sensitive Data`         | `type: default`          | `**************`       |
 
 With `custom` type, the masking function is defined by the user, by providing beginning and end indexes for masking. If the input is shorter than the end index, the rest of the string will all be masked. See the third example below.
+
 ```yaml
 transformations:
   - schema: public
@@ -327,13 +344,14 @@ transformations:
 
 **Input-Output Examples:**
 
-| Input Value              | Output Value           |
-| ------------------------ | ---------------------- |
-| `1234567812345678`       | `1234********5678`     |
-| `sensitive@example.com`  | `sens********ample.com`|
-| `sensitive`              | `sens*****`            |
+| Input Value             | Output Value            |
+| ----------------------- | ----------------------- |
+| `1234567812345678`      | `1234********5678`      |
+| `sensitive@example.com` | `sens********ample.com` |
+| `sensitive`             | `sens*****`             |
 
 If the begin index is not provided, it defaults to 0. If the end is not provided, it defaults to input length.
+
 ```yaml
 transformations:
   - schema: public
@@ -346,13 +364,14 @@ transformations:
           mask_end: "5"
 ```
 
-| Input Value              | Output Value           |
-| ------------------------ | ---------------------- |
-| `1234567812345678`       | `*****67812345678`     |
-| `sensitive@example.com`  | `*****tive@example.com`|
-| `sensitive`              | `*****tive`            |
+| Input Value             | Output Value            |
+| ----------------------- | ----------------------- |
+| `1234567812345678`      | `*****67812345678`      |
+| `sensitive@example.com` | `*****tive@example.com` |
+| `sensitive`             | `*****tive`             |
 
 Alternatively, since input length may vary, user can provide relative beginning and end indexes, as percentages of the input length.
+
 ```yaml
 transformations:
   - schema: public
@@ -365,14 +384,16 @@ transformations:
           mask_begin: "15%"
           mask_end: "85%"
 ```
-| Input Value              | Output Value           |
-| ------------------------ | ---------------------- |
-| `1234567812345678`       | `12***********678`     |
-| `sensitive@example.com`  | `sen***************com`|
-| `sensitive`              | `s******ve`            |
+
+| Input Value             | Output Value            |
+| ----------------------- | ----------------------- |
+| `1234567812345678`      | `12***********678`      |
+| `sensitive@example.com` | `sen***************com` |
+| `sensitive`             | `s******ve`             |
 
 Alternatively, user can provide unmask begin and end indexes. In that case, the specified part of the input will remain unmasked, while all the rest is masked.
-Mask and unmask parameters cannot be provided at the same time. 
+Mask and unmask parameters cannot be provided at the same time.
+
 ```yaml
 transformations:
   - schema: public
@@ -384,12 +405,12 @@ transformations:
           type: custom
           unmask_end: "3"
 ```
-| Input Value              | Output Value           |
-| ------------------------ | ---------------------- |
-| `1234567812345678`       | `123*************`     |
-| `sensitive@example.com`  | `sen******************`|
-| `sensitive`              | `sen******`            |
 
+| Input Value             | Output Value            |
+| ----------------------- | ----------------------- |
+| `1234567812345678`      | `123*************`      |
+| `sensitive@example.com` | `sen******************` |
+| `sensitive`             | `sen******`             |
 
 </details>
 
diff --git a/pg2pg.env b/pg2pg.env
index b067003..0889d50 100644
--- a/pg2pg.env
+++ b/pg2pg.env
@@ -11,6 +11,7 @@ PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS=1
 # Processor config
 PGSTREAM_TRANSFORMER_RULES_FILE="transformer_rules.yaml"
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost?sslmode=disable"
+PGSTREAM_FILTER_INCLUDE_TABLES="test"
 PGSTREAM_POSTGRES_WRITER_TARGET_URL="postgres://postgres:postgres@localhost:7654?sslmode=disable"
 PGSTREAM_POSTGRES_WRITER_BATCH_SIZE=100
 PGSTREAM_POSTGRES_WRITER_BATCH_TIMEOUT=5s
diff --git a/pg2pg.yaml b/pg2pg.yaml
index bf52bd6..73560cf 100644
--- a/pg2pg.yaml
+++ b/pg2pg.yaml
@@ -29,6 +29,9 @@ target:
 modifiers:
   injector:
     enabled: true # whether to inject pgstream metadata into the WAL events
+  filter:
+    include_tables:
+    - test
   transformations:
   - schema: public
     table: test
diff --git a/pkg/stream/config.go b/pkg/stream/config.go
index 8736a15..ca77ccd 100644
--- a/pkg/stream/config.go
+++ b/pkg/stream/config.go
@@ -10,6 +10,7 @@ import (
 	"github.com/xataio/pgstream/pkg/kafka"
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	snapshotbuilder "github.com/xataio/pgstream/pkg/wal/listener/snapshot/builder"
+	"github.com/xataio/pgstream/pkg/wal/processor/filter"
 	"github.com/xataio/pgstream/pkg/wal/processor/injector"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
 	"github.com/xataio/pgstream/pkg/wal/processor/postgres"
@@ -49,6 +50,7 @@ type ProcessorConfig struct {
 	Postgres    *PostgresProcessorConfig
 	Injector    *injector.Config
 	Transformer *transformer.Config
+	Filter      *filter.Config
 }
 
 type KafkaProcessorConfig struct {
diff --git a/pkg/stream/stream_run.go b/pkg/stream/stream_run.go
index 1aede7c..f2daa84 100644
--- a/pkg/stream/stream_run.go
+++ b/pkg/stream/stream_run.go
@@ -11,6 +11,7 @@ import (
 	kafkainstrumentation "github.com/xataio/pgstream/pkg/kafka/instrumentation"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/otel"
+	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal/checkpointer"
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	pgcheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/postgres"
@@ -20,6 +21,7 @@ import (
 	snapshotlistener "github.com/xataio/pgstream/pkg/wal/listener/snapshot"
 	snapshotbuilder "github.com/xataio/pgstream/pkg/wal/listener/snapshot/builder"
 	"github.com/xataio/pgstream/pkg/wal/processor"
+	"github.com/xataio/pgstream/pkg/wal/processor/filter"
 	"github.com/xataio/pgstream/pkg/wal/processor/injector"
 	processinstrumentation "github.com/xataio/pgstream/pkg/wal/processor/instrumentation"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
@@ -259,6 +261,20 @@ func Run(ctx context.Context, logger loglib.Logger, config *Config, instrumentat
 		processor = injector
 	}
 
+	if config.Processor.Filter != nil {
+		logger.Info("adding filtering to processor...")
+		var err error
+		processor, err = filter.New(processor, config.Processor.Filter,
+			filter.WithLogger(logger),
+			// by default we include the pgstream schema log table, since we won't
+			// be able to replicate DDL changes otherwise. This behaviour can be
+			// disabled by adding it to the exclude tables.
+			filter.WithDefaultIncludeTables([]string{schemalog.SchemaName + "." + schemalog.TableName}))
+		if err != nil {
+			return err
+		}
+	}
+
 	if processor != nil && instrumentation.IsEnabled() {
 		var err error
 		processor, err = processinstrumentation.NewProcessor(processor, instrumentation)
diff --git a/pkg/wal/processor/filter/wal_filter.go b/pkg/wal/processor/filter/wal_filter.go
new file mode 100644
index 0000000..94bc301
--- /dev/null
+++ b/pkg/wal/processor/filter/wal_filter.go
@@ -0,0 +1,201 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package filter
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	loglib "github.com/xataio/pgstream/pkg/log"
+	"github.com/xataio/pgstream/pkg/wal"
+	"github.com/xataio/pgstream/pkg/wal/processor"
+)
+
+// Filter is a processor wrapper that filter table WAL events based on the
+// configured table include/exclude lists.
+type Filter struct {
+	processor       processor.Processor
+	includeTableMap schemaTableMap
+	excludeTableMap schemaTableMap
+	logger          loglib.Logger
+}
+
+type Config struct {
+	// List of tables to allow. Tables should be schema qualified. If no schema
+	// is provided, the public schema will be assumed. Wildcards "*" are
+	// supported.
+	IncludeTables []string
+	// List of tables to skip. Tables should be schema qualified. If no schema
+	// is provided, the public schema will be assumed. Wildcards "*" are
+	// supported.
+	ExcludeTables []string
+}
+
+type schemaTableMap map[string]map[string]struct{}
+
+type Option func(*Filter)
+
+var (
+	errIncludeExcludeList     = errors.New("cannot use both include and exclude lists for table filtering")
+	errMissingFilteringConfig = errors.New("missing filtering configuration")
+	errInvalidTableName       = errors.New("invalid table name format")
+)
+
+const (
+	wildcard     = "*"
+	publicSchema = "public"
+)
+
+// New will return a filter processor wrapper that will skip WAL events as per
+// the configuration provided. Only one of include or exclude table list can be
+// provided (not both).
+func New(processor processor.Processor, cfg *Config, opts ...Option) (*Filter, error) {
+	if len(cfg.ExcludeTables) > 0 && len(cfg.IncludeTables) > 0 {
+		return nil, errIncludeExcludeList
+	}
+	if len(cfg.ExcludeTables) == 0 && len(cfg.IncludeTables) == 0 {
+		return nil, errMissingFilteringConfig
+	}
+
+	f := &Filter{
+		processor: processor,
+		logger:    loglib.NewNoopLogger(),
+	}
+
+	var err error
+	if len(cfg.IncludeTables) > 0 {
+		f.includeTableMap, err = parseTables(cfg.IncludeTables)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if len(cfg.ExcludeTables) > 0 {
+		f.excludeTableMap, err = parseTables(cfg.ExcludeTables)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	for _, opt := range opts {
+		opt(f)
+	}
+
+	return f, nil
+}
+
+func WithLogger(logger loglib.Logger) Option {
+	return func(f *Filter) {
+		f.logger = loglib.NewLogger(logger).WithFields(loglib.Fields{
+			loglib.ModuleField: "wal_filter",
+		})
+	}
+}
+
+// WithDefaultIncludeTables will use the tables on input as default include
+// table list or add them to the configured one if provided. If the filter has
+// an exclude table list configured, the default include table list won't apply,
+// since only one can be provided.
+func WithDefaultIncludeTables(tables []string) Option {
+	return func(f *Filter) {
+		if len(f.excludeTableMap) > 0 {
+			return
+		}
+		for _, table := range tables {
+			schema, table, err := parseTableName(table)
+			// should never happen
+			if err != nil {
+				panic(err)
+			}
+			f.includeTableMap.add(schema, table)
+		}
+	}
+}
+
+func (f *Filter) ProcessWALEvent(ctx context.Context, event *wal.Event) error {
+	if f.skipEvent(event) {
+		f.logger.Trace("skipping event", loglib.Fields{"schema": event.Data.Schema, "table": event.Data.Table})
+		return nil
+	}
+
+	return f.processor.ProcessWALEvent(ctx, event)
+}
+
+func (f *Filter) Name() string {
+	return f.processor.Name()
+}
+
+// skip event for table if it's not in the include table list or if it's in the exclude one
+func (f *Filter) skipEvent(event *wal.Event) bool {
+	if event == nil || event.Data == nil {
+		return false
+	}
+
+	if len(f.includeTableMap) > 0 {
+		return !f.includeTableMap.containsSchemaTable(event.Data.Schema, event.Data.Table)
+	}
+	if len(f.excludeTableMap) > 0 {
+		return f.excludeTableMap.containsSchemaTable(event.Data.Schema, event.Data.Table)
+	}
+
+	return false
+}
+
+func (t schemaTableMap) containsSchemaTable(schema, table string) bool {
+	if len(t) == 0 {
+		return false
+	}
+
+	tables := t.getSchemaTables(schema)
+	if len(tables) == 0 {
+		return false
+	}
+
+	_, found := tables[table]
+	_, wildcardFound := tables[wildcard]
+	return found || wildcardFound
+}
+
+func (t schemaTableMap) getSchemaTables(schema string) map[string]struct{} {
+	tables, found := t[schema]
+	if found {
+		return tables
+	}
+	return t[wildcard]
+}
+
+func (t schemaTableMap) add(schema, table string) {
+	_, found := t[schema]
+	if !found {
+		t[schema] = map[string]struct{}{}
+	}
+	t[schema][table] = struct{}{}
+}
+
+func parseTables(tables []string) (schemaTableMap, error) {
+	schemaTablesMap := make(schemaTableMap, len(tables))
+	for _, table := range tables {
+		schemaName, tableName, err := parseTableName(table)
+		if err != nil {
+			return nil, err
+		}
+		if _, found := schemaTablesMap[schemaName]; !found {
+			schemaTablesMap[schemaName] = make(map[string]struct{})
+		}
+		schemaTablesMap[schemaName][tableName] = struct{}{}
+	}
+	return schemaTablesMap, nil
+}
+
+func parseTableName(qualifiedTableName string) (string, string, error) {
+	parts := strings.Split(qualifiedTableName, ".")
+	switch len(parts) {
+	case 1:
+		return publicSchema, parts[0], nil
+	case 2:
+		return parts[0], parts[1], nil
+	default:
+		return "", "", errInvalidTableName
+	}
+}
diff --git a/pkg/wal/processor/injector/wal_injector.go b/pkg/wal/processor/injector/wal_injector.go
index 2c986c2..495d744 100644
--- a/pkg/wal/processor/injector/wal_injector.go
+++ b/pkg/wal/processor/injector/wal_injector.go
@@ -24,8 +24,6 @@ type Injector struct {
 	logger               loglib.Logger
 	processor            processor.Processor
 	walToLogEntryAdapter walToLogEntryAdapter
-	skipDataEvent        dataEventFilter
-	skipSchemaEvent      schemaEventFilter
 	schemaLogStore       schemalog.Store
 	idFinder             columnFinder
 	versionFinder        columnFinderWithErr
@@ -40,8 +38,6 @@ type Config struct {
 // configurable filters that allow the user of this library to have flexibility
 // when processing and injecting the wal event metadata
 type (
-	dataEventFilter     func(*wal.Data) bool
-	schemaEventFilter   func(*schemalog.LogEntry) bool
 	columnFinder        func(*schemalog.Column, *schemalog.Table) bool
 	columnFinderWithErr func(*schemalog.Column, *schemalog.Table) (bool, error)
 )
@@ -68,9 +64,6 @@ func New(cfg *Config, p processor.Processor, opts ...Option) (*Injector, error)
 		processor:            p,
 		schemaLogStore:       schemaLogStore,
 		walToLogEntryAdapter: processor.WalDataToLogEntry,
-		// by default all events are processed
-		skipDataEvent:   func(*wal.Data) bool { return false },
-		skipSchemaEvent: func(*schemalog.LogEntry) bool { return false },
 		// by default we look for the primary key to use as identity column
 		idFinder: primaryKeyFinder,
 	}
@@ -94,18 +87,6 @@ func WithVersionFinder(versionFinder columnFinderWithErr) Option {
 	}
 }
 
-func WithSkipSchemaEvent(skip schemaEventFilter) Option {
-	return func(in *Injector) {
-		in.skipSchemaEvent = skip
-	}
-}
-
-func WithSkipDataEvent(skip dataEventFilter) Option {
-	return func(in *Injector) {
-		in.skipDataEvent = skip
-	}
-}
-
 func WithLogger(l loglib.Logger) Option {
 	return func(in *Injector) {
 		in.logger = loglib.NewLogger(l).WithFields(loglib.Fields{
@@ -128,9 +109,6 @@ func (in *Injector) ProcessWALEvent(ctx context.Context, event *wal.Event) error
 	}
 
 	data := event.Data
-	if in.skipDataEvent(data) {
-		return nil
-	}
 
 	switch {
 	case isSchemaLogSchema(data.Schema):
@@ -144,10 +122,6 @@ func (in *Injector) ProcessWALEvent(ctx context.Context, event *wal.Event) error
 			return err
 		}
 
-		if in.skipSchemaEvent(logEntry) {
-			return nil
-		}
-
 		if err := in.schemaLogStore.Ack(ctx, logEntry); err != nil {
 			in.logger.Error(err, "ack schema log")
 		}
diff --git a/snapshot2pg.env b/snapshot2pg.env
index d4b7219..76138bf 100644
--- a/snapshot2pg.env
+++ b/snapshot2pg.env
@@ -1,6 +1,6 @@
 # Listener config
 PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL="postgres://postgres:postgres@localhost?sslmode=disable"
-PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
+PGSTREAM_POSTGRES_SNAPSHOT_TABLES="test"
 PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS=4
 PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS=4
 PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE=1000


Test patch:
diff --git a/cmd/config/helper_test.go b/cmd/config/helper_test.go
index 168b678..3f45ca9 100644
--- a/cmd/config/helper_test.go
+++ b/cmd/config/helper_test.go
@@ -103,4 +103,8 @@ func validateTestStreamConfig(t *testing.T, streamConfig *stream.Config) {
 	column, ok := dynParamGender["column"].(string)
 	assert.True(t, ok)
 	assert.Equal(t, "sex", column)
+
+	assert.NotNil(t, streamConfig.Processor.Filter)
+	assert.ElementsMatch(t, []string{"test", "test_schema.test", "another_schema.*"}, streamConfig.Processor.Filter.IncludeTables)
+	assert.ElementsMatch(t, []string{"excluded_test", "excluded_schema.test", "another_excluded_schema.*"}, streamConfig.Processor.Filter.ExcludeTables)
 }
diff --git a/cmd/config/test/test_config.env b/cmd/config/test/test_config.env
index 12b0114..c7ce928 100644
--- a/cmd/config/test/test_config.env
+++ b/cmd/config/test/test_config.env
@@ -80,5 +80,9 @@ PGSTREAM_WEBHOOK_NOTIFIER_CLIENT_TIMEOUT="1s"
 # Injector
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgresql://user:password@localhost:5432/mydatabase"
 
+# Filter
+PGSTREAM_FILTER_INCLUDE_TABLES="test test_schema.test another_schema.*"
+PGSTREAM_FILTER_EXCLUDE_TABLES="excluded_test excluded_schema.test another_excluded_schema.*"
+
 # Transformers
 PGSTREAM_TRANSFORMER_RULES_FILE="test/test_transformer_rules.yaml"
diff --git a/cmd/config/test/test_config.yaml b/cmd/config/test/test_config.yaml
index 97439e4..142bb5a 100644
--- a/cmd/config/test/test_config.yaml
+++ b/cmd/config/test/test_config.yaml
@@ -100,6 +100,15 @@ target:
 modifiers:
   injector:
     enabled: true # whether to inject pgstream metadata into the WAL events
+  filter: # one of include_tables or exclude_tables
+    include_tables: # list of tables for which events should be allowed. Tables should be schema qualified. If no schema is provided, the public schema will be assumed. Wildcards "*" are supported.
+     - "test"
+     - "test_schema.test"
+     - "another_schema.*"
+    exclude_tables: # list of tables for which events should be skipped. Tables should be schema qualified. If no schema is provided, the public schema will be assumed. Wildcards "*" are supported.
+     - "excluded_test"
+     - "excluded_schema.test"
+     - "another_excluded_schema.*"
   transformations:
   - schema: public
     table: test
diff --git a/pkg/wal/processor/filter/wal_filter_test.go b/pkg/wal/processor/filter/wal_filter_test.go
new file mode 100644
index 0000000..c9ff0fc
--- /dev/null
+++ b/pkg/wal/processor/filter/wal_filter_test.go
@@ -0,0 +1,323 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package filter
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/pkg/log"
+	"github.com/xataio/pgstream/pkg/wal"
+	"github.com/xataio/pgstream/pkg/wal/processor/mocks"
+)
+
+func Test_New(t *testing.T) {
+	t.Parallel()
+
+	mockProcessor := &mocks.Processor{
+		ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+			return nil
+		},
+	}
+
+	tests := []struct {
+		name         string
+		config       *Config
+		wantIncluded schemaTableMap
+		wantExcluded schemaTableMap
+		wantErr      error
+	}{
+		{
+			name: "valid included configuration",
+			config: &Config{
+				IncludeTables: []string{"users", "public.orders", "public.*", "*.*"},
+			},
+			wantIncluded: schemaTableMap{
+				"public": map[string]struct{}{
+					"*":       {},
+					"orders":  {},
+					"users":   {},
+					"default": {},
+				},
+				"*": map[string]struct{}{
+					"*": {},
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "valid excluded configuration",
+			config: &Config{
+				ExcludeTables: []string{"public.users", "public.orders"},
+			},
+			wantExcluded: schemaTableMap{
+				"public": map[string]struct{}{
+					"orders": {},
+					"users":  {},
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "both included and excluded configured",
+			config: &Config{
+				IncludeTables: []string{"public.users"},
+				ExcludeTables: []string{"public.orders"},
+			},
+			wantErr: errIncludeExcludeList,
+		},
+		{
+			name: "invalid table name in included",
+			config: &Config{
+				IncludeTables: []string{"invalid.table.name"},
+			},
+			wantErr: errInvalidTableName,
+		},
+		{
+			name: "invalid table name in excluded",
+			config: &Config{
+				ExcludeTables: []string{"invalid.table.name"},
+			},
+			wantErr: errInvalidTableName,
+		},
+		{
+			name:    "empty configuration",
+			config:  &Config{},
+			wantErr: errMissingFilteringConfig,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			filter, err := New(mockProcessor, tc.config,
+				WithDefaultIncludeTables([]string{"default"}),
+				WithLogger(log.NewNoopLogger()))
+			require.ErrorIs(t, err, tc.wantErr)
+			if filter == nil {
+				return
+			}
+			require.Equal(t, tc.wantIncluded, filter.includeTableMap)
+			require.Equal(t, tc.wantExcluded, filter.excludeTableMap)
+		})
+	}
+}
+
+func TestFilter_ProcessWALEvent(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	testSchema := "test_schema"
+	testTable := "test_table"
+	testEvent := &wal.Event{
+		Data: &wal.Data{
+			Schema: testSchema,
+			Table:  testTable,
+		},
+	}
+
+	tests := []struct {
+		name      string
+		processor *mocks.Processor
+		included  schemaTableMap
+		excluded  schemaTableMap
+		event     *wal.Event
+
+		wantErr error
+	}{
+		{
+			name:  "event is nil",
+			event: nil,
+			processor: &mocks.Processor{
+				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+					require.Nil(t, walEvent)
+					return nil
+				},
+			},
+			included: nil,
+			excluded: nil,
+			wantErr:  nil,
+		},
+		{
+			name:  "event matches included",
+			event: testEvent,
+			processor: &mocks.Processor{
+				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+					require.Equal(t, testEvent, walEvent)
+					return nil
+				},
+			},
+			included: schemaTableMap{
+				testSchema: {
+					testTable: struct{}{},
+				},
+			},
+			excluded: nil,
+			wantErr:  nil,
+		},
+		{
+			name:  "event does not match included",
+			event: testEvent,
+			processor: &mocks.Processor{
+				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+					return errors.New("ProcessWALEventFn: should not be called")
+				},
+			},
+			included: schemaTableMap{
+				"public": {
+					"orders": struct{}{},
+				},
+			},
+			excluded: nil,
+			wantErr:  nil,
+		},
+		{
+			name:  "event matches excluded",
+			event: testEvent,
+			processor: &mocks.Processor{
+				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+					return errors.New("ProcessWALEventFn: should not be called")
+				},
+			},
+			included: nil,
+			excluded: schemaTableMap{
+				testSchema: {
+					testTable: struct{}{},
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name:  "event does not match excluded",
+			event: testEvent,
+			processor: &mocks.Processor{
+				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+					require.Equal(t, testEvent, walEvent)
+					return nil
+				},
+			},
+			included: nil,
+			excluded: schemaTableMap{
+				"public": {
+					"orders": struct{}{},
+				},
+			},
+			wantErr: nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			f := &Filter{
+				logger:          log.NewNoopLogger(),
+				processor:       tc.processor,
+				excludeTableMap: tc.excluded,
+				includeTableMap: tc.included,
+			}
+
+			err := f.ProcessWALEvent(ctx, tc.event)
+			require.ErrorIs(t, err, tc.wantErr)
+		})
+	}
+}
+
+func TestSchemaTableMap_containsSchemaTable(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name         string
+		schema       string
+		table        string
+		schemaMap    schemaTableMap
+		wantContains bool
+	}{
+		{
+			name:   "table exists in schema",
+			schema: "public",
+			table:  "users",
+			schemaMap: schemaTableMap{
+				"public": {
+					"users": struct{}{},
+				},
+			},
+			wantContains: true,
+		},
+		{
+			name:   "table does not exist in schema",
+			schema: "public",
+			table:  "orders",
+			schemaMap: schemaTableMap{
+				"public": {
+					"users": struct{}{},
+				},
+			},
+			wantContains: false,
+		},
+		{
+			name:   "wildcard matches any table in schema",
+			schema: "public",
+			table:  "orders",
+			schemaMap: schemaTableMap{
+				"public": {
+					"*": struct{}{},
+				},
+			},
+			wantContains: true,
+		},
+		{
+			name:   "schema does not exist",
+			schema: "private",
+			table:  "users",
+			schemaMap: schemaTableMap{
+				"public": {
+					"users": struct{}{},
+				},
+			},
+			wantContains: false,
+		},
+		{
+			name:   "wildcard schema matches any schema",
+			schema: "private",
+			table:  "users",
+			schemaMap: schemaTableMap{
+				"*": {
+					"users": struct{}{},
+				},
+			},
+			wantContains: true,
+		},
+		{
+			name:   "wildcard schema and table match",
+			schema: "private",
+			table:  "orders",
+			schemaMap: schemaTableMap{
+				"*": {
+					"*": struct{}{},
+				},
+			},
+			wantContains: true,
+		},
+		{
+			name:         "empty schemaTableMap",
+			schema:       "public",
+			table:        "users",
+			schemaMap:    schemaTableMap{},
+			wantContains: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			contains := tc.schemaMap.containsSchemaTable(tc.schema, tc.table)
+			require.Equal(t, tc.wantContains, contains)
+		})
+	}
+}
diff --git a/pkg/wal/processor/injector/wal_injector_test.go b/pkg/wal/processor/injector/wal_injector_test.go
index 9ace417..5794509 100644
--- a/pkg/wal/processor/injector/wal_injector_test.go
+++ b/pkg/wal/processor/injector/wal_injector_test.go
@@ -22,31 +22,15 @@ func TestInjector_ProcessWALEvent(t *testing.T) {
 	testLogEntry := newTestLogEntry()
 
 	tests := []struct {
-		name            string
-		event           *wal.Event
-		store           schemalog.Store
-		adapter         walToLogEntryAdapter
-		skipDataEvent   dataEventFilter
-		skipSchemaEvent schemaEventFilter
-		idFinder        columnFinder
-		processor       processor.Processor
+		name      string
+		event     *wal.Event
+		store     schemalog.Store
+		adapter   walToLogEntryAdapter
+		idFinder  columnFinder
+		processor processor.Processor
 
 		wantErr error
 	}{
-		{
-			name:          "ok - skip schema",
-			event:         newTestDataEvent("I"),
-			skipDataEvent: func(*wal.Data) bool { return true },
-
-			wantErr: nil,
-		},
-		{
-			name:            "ok - skip log entry schema log",
-			event:           newTestSchemaChangeEvent("I"),
-			skipSchemaEvent: func(s *schemalog.LogEntry) bool { return s.SchemaName == testSchemaName },
-
-			wantErr: nil,
-		},
 		{
 			name: "ok - schema event from ignored table",
 			event: func() *wal.Event {
@@ -186,8 +170,6 @@ func TestInjector_ProcessWALEvent(t *testing.T) {
 				logger:               loglib.NewNoopLogger(),
 				processor:            tc.processor,
 				schemaLogStore:       tc.store,
-				skipDataEvent:        func(d *wal.Data) bool { return false },
-				skipSchemaEvent:      func(*schemalog.LogEntry) bool { return false },
 				idFinder:             func(c *schemalog.Column, _ *schemalog.Table) bool { return c.Name == "col-1" },
 				versionFinder:        func(c *schemalog.Column, _ *schemalog.Table) (bool, error) { return c.Name == "col-2", nil },
 				walToLogEntryAdapter: func(d *wal.Data) (*schemalog.LogEntry, error) { return testLogEntry, nil },
@@ -201,14 +183,6 @@ func TestInjector_ProcessWALEvent(t *testing.T) {
 				injector.walToLogEntryAdapter = tc.adapter
 			}
 
-			if tc.skipSchemaEvent != nil {
-				injector.skipSchemaEvent = tc.skipSchemaEvent
-			}
-
-			if tc.skipDataEvent != nil {
-				injector.skipDataEvent = tc.skipDataEvent
-			}
-
 			err := injector.ProcessWALEvent(context.Background(), tc.event)
 			require.ErrorIs(t, err, tc.wantErr)
 		})