Repository: xataio/pgstream
Pull request #137 (2025-02-07 14:22:18)

Add pg writer ddl support
This PR adds support to process DDL events on the postgres batch writer.

- Add DDL adapter to deal with the wal event->query transformations.
- Move and rename the existing adapter to DML adapter, and create an aggregator struct that handles both DDL and DML.
- Add option to fetch schema log from the schema log store by providing schema and version to allow to retrieve the previous version to compare and compute the diff.
- Make sure DDL operations are run in their own transaction in the postgres batch writer.

Commits have been split for ease of reviewing.




Patch:
diff --git a/cmd/config.go b/cmd/config.go
index 8d8e7ce..109baf6 100644
--- a/cmd/config.go
+++ b/cmd/config.go
@@ -248,6 +248,9 @@ func parsePostgresProcessorConfig() *stream.PostgresProcessorConfig {
 				MaxBatchSize:  viper.GetInt64("PGSTREAM_POSTGRES_WRITER_BATCH_SIZE"),
 				MaxQueueBytes: viper.GetInt64("PGSTREAM_POSTGRES_WRITER_MAX_QUEUE_BYTES"),
 			},
+			SchemaStore: pgschemalog.Config{
+				URL: viper.GetString("PGSTREAM_POSTGRES_WRITER_SCHEMA_STORE_URL"),
+			},
 		},
 	}
 }
diff --git a/internal/postgres/errors.go b/internal/postgres/errors.go
index 9fb29cd..bfd6004 100644
--- a/internal/postgres/errors.go
+++ b/internal/postgres/errors.go
@@ -4,6 +4,7 @@ package postgres
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
@@ -14,6 +15,14 @@ var (
 	ErrNoRows      = errors.New("no rows")
 )
 
+type ErrRelationDoesNotExist struct {
+	Details string
+}
+
+func (e *ErrRelationDoesNotExist) Error() string {
+	return fmt.Sprintf("relation does not exist: %s", e.Details)
+}
+
 func mapError(err error) error {
 	if pgconn.Timeout(err) {
 		return ErrConnTimeout
@@ -23,5 +32,14 @@ func mapError(err error) error {
 		return ErrNoRows
 	}
 
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		if pgErr.Code == "42P01" {
+			return &ErrRelationDoesNotExist{
+				Details: pgErr.Message,
+			}
+		}
+	}
+
 	return err
 }
diff --git a/pg2pg.env b/pg2pg.env
index 1c9192c..537c49b 100644
--- a/pg2pg.env
+++ b/pg2pg.env
@@ -11,3 +11,4 @@ PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost?ssl
 PGSTREAM_POSTGRES_WRITER_TARGET_URL="postgres://postgres:postgres@localhost:7654?sslmode=disable"
 PGSTREAM_POSTGRES_WRITER_BATCH_SIZE=100
 PGSTREAM_POSTGRES_WRITER_BATCH_TIMEOUT=5s
+PGSTREAM_POSTGRES_WRITER_SCHEMA_STORE_URL="postgres://postgres:postgres@localhost?sslmode=disable"
diff --git a/pkg/schemalog/instrumentation/instrumented_store.go b/pkg/schemalog/instrumentation/instrumented_store.go
index ceb9d74..c8c75da 100644
--- a/pkg/schemalog/instrumentation/instrumented_store.go
+++ b/pkg/schemalog/instrumentation/instrumented_store.go
@@ -33,10 +33,16 @@ func (s *Store) Insert(ctx context.Context, schemaName string) (le *schemalog.Lo
 	return s.inner.Insert(ctx, schemaName)
 }
 
-func (s *Store) Fetch(ctx context.Context, schemaName string, acked bool) (le *schemalog.LogEntry, err error) {
+func (s *Store) FetchLast(ctx context.Context, schemaName string, acked bool) (le *schemalog.LogEntry, err error) {
+	ctx, span := otel.StartSpan(ctx, s.tracer, "schemalogstore.FetchLast", trace.WithAttributes(attribute.String("schema", schemaName)))
+	defer otel.CloseSpan(span, err)
+	return s.inner.FetchLast(ctx, schemaName, acked)
+}
+
+func (s *Store) Fetch(ctx context.Context, schemaName string, version int) (le *schemalog.LogEntry, err error) {
 	ctx, span := otel.StartSpan(ctx, s.tracer, "schemalogstore.Fetch", trace.WithAttributes(attribute.String("schema", schemaName)))
 	defer otel.CloseSpan(span, err)
-	return s.inner.Fetch(ctx, schemaName, acked)
+	return s.inner.Fetch(ctx, schemaName, version)
 }
 
 func (s *Store) Ack(ctx context.Context, le *schemalog.LogEntry) (err error) {
diff --git a/pkg/schemalog/mocks/store_mock.go b/pkg/schemalog/mocks/store_mock.go
index 8cb8e6c..37a6e09 100644
--- a/pkg/schemalog/mocks/store_mock.go
+++ b/pkg/schemalog/mocks/store_mock.go
@@ -11,7 +11,8 @@ import (
 
 type Store struct {
 	InsertFn    func(ctx context.Context, schemaName string) (*schemalog.LogEntry, error)
-	FetchFn     func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error)
+	FetchLastFn func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error)
+	FetchFn     func(ctx context.Context, schemaName string, version int) (*schemalog.LogEntry, error)
 	AckFn       func(ctx context.Context, le *schemalog.LogEntry) error
 	CloseFn     func() error
 	insertCalls uint64
@@ -26,9 +27,14 @@ func (m *Store) Insert(ctx context.Context, schemaName string) (*schemalog.LogEn
 	return m.InsertFn(ctx, schemaName)
 }
 
-func (m *Store) Fetch(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+func (m *Store) FetchLast(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 	atomic.AddUint64(&m.fetchCalls, 1)
-	return m.FetchFn(ctx, schemaName, ackedOnly)
+	return m.FetchLastFn(ctx, schemaName, ackedOnly)
+}
+
+func (m *Store) Fetch(ctx context.Context, schemaName string, version int) (*schemalog.LogEntry, error) {
+	atomic.AddUint64(&m.fetchCalls, 1)
+	return m.FetchFn(ctx, schemaName, version)
 }
 
 func (m *Store) Ack(ctx context.Context, le *schemalog.LogEntry) error {
diff --git a/pkg/schemalog/postgres/pg_schemalog_store.go b/pkg/schemalog/postgres/pg_schemalog_store.go
index 9aca68e..2be3314 100644
--- a/pkg/schemalog/postgres/pg_schemalog_store.go
+++ b/pkg/schemalog/postgres/pg_schemalog_store.go
@@ -64,9 +64,9 @@ func (s *Store) Insert(ctx context.Context, schemaName string) (*schemalog.LogEn
 	return l, nil
 }
 
-// Fetch retrieves the latest schema log entry for schemaName. If ackedOnly is set, the function returns the last acked
+// FetchLast retrieves the latest schema log entry for schemaName. If ackedOnly is set, the function returns the last acked
 // entry. If ackedOnly is NOT set, the function returns the latest entry no matter the acked flag.
-func (s *Store) Fetch(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+func (s *Store) FetchLast(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 	ackCondition := ""
 	if ackedOnly {
 		ackCondition = "and acked"
@@ -82,6 +82,19 @@ func (s *Store) Fetch(ctx context.Context, schemaName string, ackedOnly bool) (*
 	return l, nil
 }
 
+// Fetch retrieves the log entry for the schema and version provided.
+func (s *Store) Fetch(ctx context.Context, schemaName string, version int) (*schemalog.LogEntry, error) {
+	sql := fmt.Sprintf(`select id, version, schema_name, schema, created_at, acked from %s where schema_name = $1 and version = $2`, s.table())
+
+	l := &schemalog.LogEntry{}
+	err := s.querier.QueryRow(ctx, sql, schemaName, version).Scan(&l.ID, &l.Version, &l.SchemaName, &l.Schema, &l.CreatedAt, &l.Acked)
+	if err != nil {
+		return nil, mapError(fmt.Errorf("error fetching schema log for schema %s and version %d: %w", schemaName, version, err))
+	}
+
+	return l, nil
+}
+
 func (s *Store) Ack(ctx context.Context, logEntry *schemalog.LogEntry) error {
 	sql := fmt.Sprintf(`update %s set acked = true where id = $1 and schema_name = $2`, s.table())
 	_, err := s.querier.Exec(ctx, sql, logEntry.ID.String(), logEntry.SchemaName)
diff --git a/pkg/schemalog/store.go b/pkg/schemalog/store.go
index 4e2e7e9..fdda8cc 100644
--- a/pkg/schemalog/store.go
+++ b/pkg/schemalog/store.go
@@ -9,7 +9,8 @@ import (
 
 type Store interface {
 	Insert(ctx context.Context, schemaName string) (*LogEntry, error)
-	Fetch(ctx context.Context, schemaName string, ackedOnly bool) (*LogEntry, error)
+	FetchLast(ctx context.Context, schemaName string, ackedOnly bool) (*LogEntry, error)
+	Fetch(ctx context.Context, schemaName string, version int) (*LogEntry, error)
 	Ack(ctx context.Context, le *LogEntry) error
 	Close() error
 }
diff --git a/pkg/schemalog/store_cache.go b/pkg/schemalog/store_cache.go
index 0d5844b..537efbb 100644
--- a/pkg/schemalog/store_cache.go
+++ b/pkg/schemalog/store_cache.go
@@ -26,13 +26,13 @@ func (s *StoreCache) Insert(ctx context.Context, schemaName string) (*LogEntry,
 	return s.store.Insert(ctx, schemaName)
 }
 
-func (s *StoreCache) Fetch(ctx context.Context, schemaName string, ackedOnly bool) (*LogEntry, error) {
+func (s *StoreCache) FetchLast(ctx context.Context, schemaName string, ackedOnly bool) (*LogEntry, error) {
 	logEntry := s.cache[schemaName]
 	if logEntry == nil {
 		var err error
-		logEntry, err = s.store.Fetch(ctx, schemaName, ackedOnly)
+		logEntry, err = s.store.FetchLast(ctx, schemaName, ackedOnly)
 		if err != nil {
-			return nil, fmt.Errorf("store cache fetch: %w", err)
+			return nil, fmt.Errorf("store cache fetch last schema log: %w", err)
 		}
 		s.cache[schemaName] = logEntry
 	}
@@ -40,6 +40,10 @@ func (s *StoreCache) Fetch(ctx context.Context, schemaName string, ackedOnly boo
 	return logEntry, nil
 }
 
+func (s *StoreCache) Fetch(ctx context.Context, schemaName string, version int) (*LogEntry, error) {
+	return s.store.Fetch(ctx, schemaName, version)
+}
+
 func (s *StoreCache) Ack(ctx context.Context, entry *LogEntry) error {
 	s.cache[entry.SchemaName] = entry
 	if err := s.store.Ack(ctx, entry); err != nil {
diff --git a/pkg/wal/processor/injector/wal_injector.go b/pkg/wal/processor/injector/wal_injector.go
index 65305be..2c986c2 100644
--- a/pkg/wal/processor/injector/wal_injector.go
+++ b/pkg/wal/processor/injector/wal_injector.go
@@ -193,7 +193,7 @@ func (in *Injector) inject(ctx context.Context, data *wal.Data) error {
 		return nil
 	}
 
-	logEntry, err := in.schemaLogStore.Fetch(ctx, data.Schema, true)
+	logEntry, err := in.schemaLogStore.FetchLast(ctx, data.Schema, true)
 	if err != nil {
 		// if schema does NOT exist in the log, skip the event injection.
 		if errors.Is(err, schemalog.ErrNoRows) {
diff --git a/pkg/wal/processor/postgres/config.go b/pkg/wal/processor/postgres/config.go
index 54c8414..7de669a 100644
--- a/pkg/wal/processor/postgres/config.go
+++ b/pkg/wal/processor/postgres/config.go
@@ -3,10 +3,12 @@
 package postgres
 
 import (
+	schemalogpg "github.com/xataio/pgstream/pkg/schemalog/postgres"
 	"github.com/xataio/pgstream/pkg/wal/processor/batch"
 )
 
 type Config struct {
 	URL         string
 	BatchConfig batch.Config
+	SchemaStore schemalogpg.Config
 }
diff --git a/pkg/wal/processor/postgres/postgres_batch_writer.go b/pkg/wal/processor/postgres/postgres_batch_writer.go
index dbc533b..a3844f2 100644
--- a/pkg/wal/processor/postgres/postgres_batch_writer.go
+++ b/pkg/wal/processor/postgres/postgres_batch_writer.go
@@ -4,11 +4,14 @@ package postgres
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"runtime/debug"
 
 	pglib "github.com/xataio/pgstream/internal/postgres"
 	loglib "github.com/xataio/pgstream/pkg/log"
+	"github.com/xataio/pgstream/pkg/schemalog"
+	schemalogpg "github.com/xataio/pgstream/pkg/schemalog/postgres"
 	"github.com/xataio/pgstream/pkg/wal"
 	"github.com/xataio/pgstream/pkg/wal/checkpointer"
 	"github.com/xataio/pgstream/pkg/wal/processor"
@@ -44,10 +47,17 @@ func NewBatchWriter(ctx context.Context, config *Config, opts ...Option) (*Batch
 		return nil, err
 	}
 
+	var schemaLogStore schemalog.Store
+	schemaLogStore, err = schemalogpg.NewStore(ctx, config.SchemaStore)
+	if err != nil {
+		return nil, fmt.Errorf("create schema log postgres store: %w", err)
+	}
+	schemaLogStore = schemalog.NewStoreCache(schemaLogStore)
+
 	w := &BatchWriter{
 		logger:  loglib.NewNoopLogger(),
 		pgConn:  pgConn,
-		adapter: &adapter{},
+		adapter: newAdapter(schemaLogStore),
 	}
 
 	for _, opt := range opts {
@@ -98,13 +108,20 @@ func (w *BatchWriter) ProcessWALEvent(ctx context.Context, walEvent *wal.Event)
 		}
 	}()
 
-	query, err := w.adapter.walEventToQuery(walEvent)
+	queries, err := w.adapter.walEventToQueries(ctx, walEvent)
 	if err != nil {
 		return err
 	}
 
-	msg := batch.NewWALMessage(query, walEvent.CommitPosition)
-	return w.batchSender.AddToBatch(ctx, msg)
+	for _, q := range queries {
+		w.logger.Debug("batching query", loglib.Fields{"query": q})
+		msg := batch.NewWALMessage(q, walEvent.CommitPosition)
+		if err := w.batchSender.AddToBatch(ctx, msg); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func (w *BatchWriter) Name() string {
@@ -117,8 +134,49 @@ func (w *BatchWriter) Close() error {
 }
 
 func (w *BatchWriter) sendBatch(ctx context.Context, batch *batch.Batch[*query]) error {
+	// we'll mostly process DML queries, so pre-allocate the max size
+	dmlQueries := make([]*query, 0, len(batch.GetMessages()))
+	for _, q := range batch.GetMessages() {
+		if !q.isDDL {
+			dmlQueries = append(dmlQueries, q)
+			continue
+		}
+
+		// flush any previous DML queries before running the DDL query to ensure
+		// they are run in their own separate transaction
+		if err := w.flushQueries(ctx, dmlQueries); err != nil {
+			w.logger.Error(err, "flushing DML queries")
+			return err
+		}
+
+		if _, err := w.pgConn.Exec(ctx, q.sql, q.args...); err != nil {
+			w.logger.Error(err, "running DDL query", loglib.Fields{"query_sql": q.sql, "query_args": q.args})
+			var errRelationDoesNotExist *pglib.ErrRelationDoesNotExist
+			if errors.As(err, &errRelationDoesNotExist) {
+				continue
+			}
+			return err
+		}
+	}
+
+	if err := w.flushQueries(ctx, dmlQueries); err != nil {
+		w.logger.Error(err, "flushing DML queries")
+		return err
+	}
+
+	if w.checkpointer != nil {
+		return w.checkpointer(ctx, batch.GetCommitPositions())
+	}
+
+	return nil
+}
+
+func (w *BatchWriter) flushQueries(ctx context.Context, queries []*query) error {
+	if len(queries) == 0 {
+		return nil
+	}
 	err := w.pgConn.ExecInTx(ctx, func(tx pglib.Tx) error {
-		for _, q := range batch.GetMessages() {
+		for _, q := range queries {
 			if _, err := tx.Exec(ctx, q.sql, q.args...); err != nil {
 				return err
 			}
@@ -129,9 +187,6 @@ func (w *BatchWriter) sendBatch(ctx context.Context, batch *batch.Batch[*query])
 		return err
 	}
 
-	if w.checkpointer != nil {
-		return w.checkpointer(ctx, batch.GetCommitPositions())
-	}
-
+	queries = queries[:0]
 	return nil
 }
diff --git a/pkg/wal/processor/postgres/postgres_query_msg.go b/pkg/wal/processor/postgres/postgres_query_msg.go
index ff85365..50345e9 100644
--- a/pkg/wal/processor/postgres/postgres_query_msg.go
+++ b/pkg/wal/processor/postgres/postgres_query_msg.go
@@ -3,8 +3,9 @@
 package postgres
 
 type query struct {
-	sql  string
-	args []any
+	sql   string
+	args  []any
+	isDDL bool
 }
 
 // size returns the size of the message sql query (does not include the
diff --git a/pkg/wal/processor/postgres/postgres_wal_adapter.go b/pkg/wal/processor/postgres/postgres_wal_adapter.go
index c31bff9..7a59dcf 100644
--- a/pkg/wal/processor/postgres/postgres_wal_adapter.go
+++ b/pkg/wal/processor/postgres/postgres_wal_adapter.go
@@ -3,146 +3,36 @@
 package postgres
 
 import (
-	"fmt"
-	"slices"
-	"strings"
+	"context"
 
-	"github.com/lib/pq"
 	"github.com/xataio/pgstream/pkg/wal"
 	"github.com/xataio/pgstream/pkg/wal/processor"
 )
 
 type walAdapter interface {
-	walEventToQuery(e *wal.Event) (*query, error)
+	walEventToQueries(ctx context.Context, e *wal.Event) ([]*query, error)
 }
 
-type adapter struct{}
-
-func (a *adapter) walEventToQuery(e *wal.Event) (*query, error) {
-	// TODO: support DDL parsing
-	if e.Data == nil || processor.IsSchemaLogEvent(e.Data) {
-		return &query{}, nil
-	}
-
-	// DML events
-	switch e.Data.Action {
-	case "I", "U", "D", "T":
-		return a.walDataToQuery(e.Data), nil
-	default:
-		// unsupported actions
-		return &query{}, nil
-	}
-}
-
-func (a *adapter) walDataToQuery(d *wal.Data) *query {
-	switch d.Action {
-	case "T":
-		return a.buildTruncateQuery(d)
-	case "D":
-		return a.buildDeleteQuery(d)
-	case "I":
-		return a.buildInsertQuery(d)
-	case "U":
-		return a.buildUpdateQuery(d)
-	default:
-		return nil
-	}
-}
-
-func (a *adapter) buildTruncateQuery(d *wal.Data) *query {
-	return &query{
-		sql: fmt.Sprintf("TRUNCATE %s", quotedTableName(d.Schema, d.Table)),
-	}
-}
-
-func (a *adapter) buildDeleteQuery(d *wal.Data) *query {
-	primaryKeyCols := a.extractPrimaryKeyColumns(d.Metadata.InternalColIDs, d.Identity)
-	whereQuery, whereValues := a.buildWhereQuery(primaryKeyCols, 0)
-	return &query{
-		sql:  fmt.Sprintf("DELETE FROM %s %s", quotedTableName(d.Schema, d.Table), whereQuery),
-		args: whereValues,
-	}
-}
-
-func (a *adapter) buildInsertQuery(d *wal.Data) *query {
-	names := make([]string, 0, len(d.Columns))
-	values := make([]any, 0, len(d.Columns))
-	placeholders := make([]string, 0, len(d.Columns))
-	for i, col := range d.Columns {
-		names = append(names, col.Name)
-		values = append(values, col.Value)
-		placeholders = append(placeholders, fmt.Sprintf("$%d", i+1))
-	}
-
-	return &query{
-		sql: fmt.Sprintf("INSERT INTO %s(%s) VALUES(%s) %s",
-			quotedTableName(d.Schema, d.Table), strings.Join(names, ", "),
-			strings.Join(placeholders, ", "),
-			a.buildOnConflictQuery(d)),
-		args: values,
-	}
-}
-
-func (a *adapter) buildUpdateQuery(d *wal.Data) *query {
-	setQuery, setValues := a.buildSetQuery(d.Columns)
-	primaryKeyCols := a.extractPrimaryKeyColumns(d.Metadata.InternalColIDs, d.Columns)
-	whereQuery, whereValues := a.buildWhereQuery(primaryKeyCols, len(d.Columns))
-	return &query{
-		sql:  fmt.Sprintf("UPDATE %s %s %s", quotedTableName(d.Schema, d.Table), setQuery, whereQuery),
-		args: append(setValues, whereValues...),
-	}
+type adapter struct {
+	dmlAdapter *dmlAdapter
+	ddlAdapter *ddlAdapter
 }
 
-func (a *adapter) buildWhereQuery(cols []wal.Column, placeholderOffset int) (string, []any) {
-	whereQuery := "WHERE"
-	whereValues := make([]any, 0, len(cols))
-	for i, c := range cols {
-		if i != 0 {
-			whereQuery = fmt.Sprintf("%s AND", whereQuery)
-		}
-		whereQuery = fmt.Sprintf("%s %s = $%d", whereQuery, c.Name, i+placeholderOffset+1)
-		whereValues = append(whereValues, c.Value)
+func newAdapter(schemaQuerier schemalogQuerier) *adapter {
+	return &adapter{
+		dmlAdapter: &dmlAdapter{},
+		ddlAdapter: newDDLAdapter(schemaQuerier),
 	}
-	return whereQuery, whereValues
 }
 
-func (a *adapter) buildSetQuery(cols []wal.Column) (string, []any) {
-	setQuery := "SET"
-	setValues := make([]any, 0, len(cols))
-	for i, c := range cols {
-		if i != 0 {
-			setQuery = fmt.Sprintf("%s,", setQuery)
-		}
-		setQuery = fmt.Sprintf("%s %s = $%d", setQuery, c.Name, i+1)
-		setValues = append(setValues, c.Value)
+func (a *adapter) walEventToQueries(ctx context.Context, e *wal.Event) ([]*query, error) {
+	if e.Data == nil {
+		return []*query{{}}, nil
 	}
-	return setQuery, setValues
-}
 
-func (a *adapter) buildOnConflictQuery(d *wal.Data) string {
-	primaryKeyCols := a.extractPrimaryKeyColumns(d.Metadata.InternalColIDs, d.Columns)
-	conflictCols := []string{}
-	for _, col := range primaryKeyCols {
-		conflictCols = append(conflictCols, col.Name)
-	}
-	if len(conflictCols) == 0 {
-		return ""
+	if processor.IsSchemaLogEvent(e.Data) {
+		return a.ddlAdapter.walDataToQueries(ctx, e.Data)
 	}
-	return fmt.Sprintf("ON CONFLICT (%s) DO NOTHING", strings.Join(conflictCols, ", "))
-}
-
-func (a *adapter) extractPrimaryKeyColumns(colIDs []string, cols []wal.Column) []wal.Column {
-	primaryKeyColumns := make([]wal.Column, 0, len(colIDs))
-	for _, col := range cols {
-		if !slices.Contains(colIDs, col.ID) {
-			continue
-		}
-		primaryKeyColumns = append(primaryKeyColumns, col)
-	}
-
-	return primaryKeyColumns
-}
 
-func quotedTableName(schemaName, tableName string) string {
-	return fmt.Sprintf("%s.%s", pq.QuoteIdentifier(schemaName), pq.QuoteIdentifier(tableName))
+	return []*query{a.dmlAdapter.walDataToQuery(e.Data)}, nil
 }
diff --git a/pkg/wal/processor/postgres/postgres_wal_ddl_adapter.go b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter.go
new file mode 100644
index 0000000..7fc09e7
--- /dev/null
+++ b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter.go
@@ -0,0 +1,231 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+
+	"github.com/xataio/pgstream/pkg/schemalog"
+	"github.com/xataio/pgstream/pkg/wal"
+	"github.com/xataio/pgstream/pkg/wal/processor"
+)
+
+type ddlAdapter struct {
+	schemalogQuerier schemalogQuerier
+	schemaDiffer     schemaDiffer
+	logEntryAdapter  logEntryAdapter
+}
+
+type schemalogQuerier interface {
+	Fetch(ctx context.Context, schemaName string, version int) (*schemalog.LogEntry, error)
+}
+
+type schemaDiffer func(old, new *schemalog.LogEntry) *schemalog.Diff
+
+type logEntryAdapter func(*wal.Data) (*schemalog.LogEntry, error)
+
+func newDDLAdapter(querier schemalogQuerier) *ddlAdapter {
+	return &ddlAdapter{
+		schemalogQuerier: querier,
+		schemaDiffer:     schemalog.ComputeSchemaDiff,
+		logEntryAdapter:  processor.WalDataToLogEntry,
+	}
+}
+
+func (a *ddlAdapter) walDataToQueries(ctx context.Context, d *wal.Data) ([]*query, error) {
+	newSchemaLog, err := a.logEntryAdapter(d)
+	if err != nil {
+		return nil, err
+	}
+
+	previousSchemaLog, err := a.schemalogQuerier.Fetch(ctx, newSchemaLog.SchemaName, int(newSchemaLog.Version)-1)
+	if err != nil && !errors.Is(err, schemalog.ErrNoRows) {
+		return nil, fmt.Errorf("fetching existing schema log entry: %w", err)
+	}
+
+	diff := a.schemaDiffer(previousSchemaLog, newSchemaLog)
+
+	return a.schemaDiffToQueries(newSchemaLog.SchemaName, diff)
+}
+
+func (a *ddlAdapter) schemaDiffToQueries(schemaName string, diff *schemalog.Diff) ([]*query, error) {
+	if diff.IsEmpty() {
+		return []*query{}, nil
+	}
+
+	queries := []*query{}
+	for _, table := range diff.TablesRemoved {
+		dropQuery := fmt.Sprintf("DROP TABLE IF EXISTS %s", quotedTableName(schemaName, table.Name))
+		queries = append(queries, a.newDDLQuery(dropQuery))
+	}
+
+	for _, table := range diff.TablesAdded {
+		queries = append(queries, a.buildCreateTableQuery(schemaName, table))
+	}
+
+	for _, tableDiff := range diff.TablesChanged {
+		queries = append(queries, a.buildAlterTableQueries(schemaName, tableDiff)...)
+	}
+
+	return queries, nil
+}
+
+func (a *ddlAdapter) buildCreateTableQuery(schemaName string, table schemalog.Table) *query {
+	createQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (", quotedTableName(schemaName, table.Name))
+	uniqueConstraints := make([]string, 0, len(table.Columns))
+	for _, col := range table.Columns {
+		createQuery = fmt.Sprintf("%s %s,\n", createQuery, a.buildColumnDefinition(&col))
+		// if there's a unique constraint associated to the column, and it's not
+		// the primary key, explicitly add it
+		if uniqueConstraint := a.buildUniqueConstraint(col); uniqueConstraint != "" && !slices.Contains(table.PrimaryKeyColumns, col.Name) {
+			uniqueConstraints = append(uniqueConstraints, uniqueConstraint)
+		}
+	}
+
+	for _, constraint := range uniqueConstraints {
+		createQuery = fmt.Sprintf("%s %s,\n", createQuery, constraint)
+	}
+
+	if len(table.PrimaryKeyColumns) > 0 {
+		createQuery = fmt.Sprintf("%s PRIMARY KEY (%s)\n", createQuery, strings.Join(table.PrimaryKeyColumns, ","))
+	}
+
+	createQuery = fmt.Sprintf("%s)", createQuery)
+
+	return a.newDDLQuery(createQuery)
+}
+
+func (a *ddlAdapter) buildColumnDefinition(column *schemalog.Column) string {
+	colDefinition := fmt.Sprintf("%s %s", column.Name, column.DataType)
+	if !column.Nullable {
+		colDefinition = fmt.Sprintf("%s NOT NULL", colDefinition)
+	}
+	// do not set default values with sequences since they will differ between
+	// source/target. Keep source database as source of truth.
+	if column.DefaultValue != nil && !strings.Contains(*column.DefaultValue, "seq") {
+		colDefinition = fmt.Sprintf("%s DEFAULT %s", colDefinition, *column.DefaultValue)
+	}
+
+	return colDefinition
+}
+
+func (a *ddlAdapter) buildUniqueConstraint(column schemalog.Column) string {
+	if column.Unique {
+		return fmt.Sprintf("UNIQUE (%s)", column.Name)
+	}
+	return ""
+}
+
+func (a *ddlAdapter) buildAlterTableQueries(schemaName string, tableDiff schemalog.TableDiff) []*query {
+	if tableDiff.IsEmpty() {
+		return []*query{}
+	}
+
+	queries := []*query{}
+	if tableDiff.TableNameChange != nil {
+		alterQuery := fmt.Sprintf("ALTER TABLE %s RENAME TO %s",
+			quotedTableName(schemaName, tableDiff.TableNameChange.Old),
+			tableDiff.TableNameChange.New,
+		)
+		queries = append(queries, a.newDDLQuery(alterQuery))
+	}
+
+	for _, col := range tableDiff.ColumnsRemoved {
+		alterQuery := fmt.Sprintf("ALTER TABLE %s DROP COLUMN %s", quotedTableName(schemaName, tableDiff.TableName), col.Name)
+		queries = append(queries, a.newDDLQuery(alterQuery))
+	}
+
+	for _, col := range tableDiff.ColumnsAdded {
+		alterQuery := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s", quotedTableName(schemaName, tableDiff.TableName), a.buildColumnDefinition(&col))
+		queries = append(queries, a.newDDLQuery(alterQuery))
+	}
+
+	for _, colDiff := range tableDiff.ColumnsChanged {
+		alterQueries := a.buildAlterColumnQueries(schemaName, tableDiff.TableName, &colDiff)
+		queries = append(queries, alterQueries...)
+	}
+
+	return queries
+}
+
+func (a *ddlAdapter) buildAlterColumnQueries(schemaName, tableName string, columnDiff *schemalog.ColumnDiff) []*query {
+	if columnDiff.IsEmpty() {
+		return []*query{}
+	}
+
+	queries := []*query{}
+	if columnDiff.NameChange != nil {
+		alterQuery := fmt.Sprintf("ALTER TABLE %s RENAME COLUMN %s TO %s",
+			quotedTableName(schemaName, tableName),
+			columnDiff.NameChange.Old,
+			columnDiff.NameChange.New,
+		)
+		queries = append(queries, a.newDDLQuery(alterQuery))
+	}
+
+	if columnDiff.TypeChange != nil {
+		alterQuery := fmt.Sprintf("ALTER TABLE %s ALTER COLUMN %s TYPE %s",
+			quotedTableName(schemaName, tableName),
+			columnDiff.ColumnName,
+			columnDiff.TypeChange.New,
+		)
+		queries = append(queries, a.newDDLQuery(alterQuery))
+	}
+
+	if columnDiff.NullChange != nil {
+		alterQuery := ""
+		switch {
+		// from not nullable to nullable
+		case columnDiff.NullChange.New:
+			alterQuery = fmt.Sprintf("ALTER TABLE %s ALTER COLUMN %s DROP NOT NULL",
+				quotedTableName(schemaName, tableName),
+				columnDiff.ColumnName,
+			)
+		default:
+			// from nullable to not nullable
+			alterQuery = fmt.Sprintf("ALTER TABLE %s ALTER COLUMN %s SET NOT NULL",
+				quotedTableName(schemaName, tableName),
+				columnDiff.ColumnName,
+			)
+		}
+		queries = append(queries, a.newDDLQuery(alterQuery))
+	}
+
+	if columnDiff.DefaultChange != nil {
+		alterQuery := ""
+		switch {
+		// removing the default
+		case columnDiff.DefaultChange.New == nil:
+			alterQuery = fmt.Sprintf("ALTER TABLE %s ALTER COLUMN %s DROP DEFAULT",
+				quotedTableName(schemaName, tableName),
+				columnDiff.ColumnName,
+			)
+		default:
+			// do not set default values with sequences since they will differ between
+			// source/target. Keep source database as source of truth.
+			if !strings.Contains(*columnDiff.DefaultChange.New, "seq") {
+				alterQuery = fmt.Sprintf("ALTER TABLE %s ALTER COLUMN %s SET DEFAULT %s",
+					quotedTableName(schemaName, tableName),
+					columnDiff.ColumnName,
+					*columnDiff.DefaultChange.New,
+				)
+			}
+		}
+		queries = append(queries, a.newDDLQuery(alterQuery))
+	}
+
+	// TODO: add support for unique constraint changes
+
+	return queries
+}
+
+func (a *ddlAdapter) newDDLQuery(sql string) *query {
+	return &query{
+		sql:   sql,
+		isDDL: true,
+	}
+}
diff --git a/pkg/wal/processor/postgres/postgres_wal_dml_adapter.go b/pkg/wal/processor/postgres/postgres_wal_dml_adapter.go
new file mode 100644
index 0000000..41dbf74
--- /dev/null
+++ b/pkg/wal/processor/postgres/postgres_wal_dml_adapter.go
@@ -0,0 +1,127 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"fmt"
+	"slices"
+	"strings"
+
+	"github.com/lib/pq"
+	"github.com/xataio/pgstream/pkg/wal"
+)
+
+type dmlAdapter struct{}
+
+func (a *dmlAdapter) walDataToQuery(d *wal.Data) *query {
+	switch d.Action {
+	case "T":
+		return a.buildTruncateQuery(d)
+	case "D":
+		return a.buildDeleteQuery(d)
+	case "I":
+		return a.buildInsertQuery(d)
+	case "U":
+		return a.buildUpdateQuery(d)
+	default:
+		return &query{}
+	}
+}
+
+func (a *dmlAdapter) buildTruncateQuery(d *wal.Data) *query {
+	return &query{
+		sql: fmt.Sprintf("TRUNCATE %s", quotedTableName(d.Schema, d.Table)),
+	}
+}
+
+func (a *dmlAdapter) buildDeleteQuery(d *wal.Data) *query {
+	primaryKeyCols := a.extractPrimaryKeyColumns(d.Metadata.InternalColIDs, d.Identity)
+	whereQuery, whereValues := a.buildWhereQuery(primaryKeyCols, 0)
+	return &query{
+		sql:  fmt.Sprintf("DELETE FROM %s %s", quotedTableName(d.Schema, d.Table), whereQuery),
+		args: whereValues,
+	}
+}
+
+func (a *dmlAdapter) buildInsertQuery(d *wal.Data) *query {
+	names := make([]string, 0, len(d.Columns))
+	values := make([]any, 0, len(d.Columns))
+	placeholders := make([]string, 0, len(d.Columns))
+	for i, col := range d.Columns {
+		names = append(names, col.Name)
+		values = append(values, col.Value)
+		placeholders = append(placeholders, fmt.Sprintf("$%d", i+1))
+	}
+
+	return &query{
+		sql: fmt.Sprintf("INSERT INTO %s(%s) VALUES(%s) %s",
+			quotedTableName(d.Schema, d.Table), strings.Join(names, ", "),
+			strings.Join(placeholders, ", "),
+			a.buildOnConflictQuery(d)),
+		args: values,
+	}
+}
+
+func (a *dmlAdapter) buildUpdateQuery(d *wal.Data) *query {
+	setQuery, setValues := a.buildSetQuery(d.Columns)
+	primaryKeyCols := a.extractPrimaryKeyColumns(d.Metadata.InternalColIDs, d.Columns)
+	whereQuery, whereValues := a.buildWhereQuery(primaryKeyCols, len(d.Columns))
+	return &query{
+		sql:  fmt.Sprintf("UPDATE %s %s %s", quotedTableName(d.Schema, d.Table), setQuery, whereQuery),
+		args: append(setValues, whereValues...),
+	}
+}
+
+func (a *dmlAdapter) buildWhereQuery(cols []wal.Column, placeholderOffset int) (string, []any) {
+	whereQuery := "WHERE"
+	whereValues := make([]any, 0, len(cols))
+	for i, c := range cols {
+		if i != 0 {
+			whereQuery = fmt.Sprintf("%s AND", whereQuery)
+		}
+		whereQuery = fmt.Sprintf("%s %s = $%d", whereQuery, c.Name, i+placeholderOffset+1)
+		whereValues = append(whereValues, c.Value)
+	}
+	return whereQuery, whereValues
+}
+
+func (a *dmlAdapter) buildSetQuery(cols []wal.Column) (string, []any) {
+	setQuery := "SET"
+	setValues := make([]any, 0, len(cols))
+	for i, c := range cols {
+		if i != 0 {
+			setQuery = fmt.Sprintf("%s,", setQuery)
+		}
+		setQuery = fmt.Sprintf("%s %s = $%d", setQuery, c.Name, i+1)
+		setValues = append(setValues, c.Value)
+	}
+	return setQuery, setValues
+}
+
+func (a *dmlAdapter) buildOnConflictQuery(d *wal.Data) string {
+	primaryKeyCols := a.extractPrimaryKeyColumns(d.Metadata.InternalColIDs, d.Columns)
+	conflictCols := []string{}
+	for _, col := range primaryKeyCols {
+		conflictCols = append(conflictCols, col.Name)
+	}
+	if len(conflictCols) == 0 {
+		return ""
+	}
+	return fmt.Sprintf("ON CONFLICT (%s) DO NOTHING", strings.Join(conflictCols, ", "))
+}
+
+func (a *dmlAdapter) extractPrimaryKeyColumns(colIDs []string, cols []wal.Column) []wal.Column {
+	primaryKeyColumns := make([]wal.Column, 0, len(colIDs))
+	for _, col := range cols {
+		if !slices.Contains(colIDs, col.ID) {
+			continue
+		}
+		primaryKeyColumns = append(primaryKeyColumns, col)
+	}
+
+	return primaryKeyColumns
+}
+
+func quotedTableName(schemaName, tableName string) string {
+	return fmt.Sprintf("%s.%s", pq.QuoteIdentifier(schemaName), pq.QuoteIdentifier(tableName))
+}


Test patch:
diff --git a/pkg/schemalog/postgres/pg_schemalog_store_test.go b/pkg/schemalog/postgres/pg_schemalog_store_test.go
index 02c2fb9..f695d74 100644
--- a/pkg/schemalog/postgres/pg_schemalog_store_test.go
+++ b/pkg/schemalog/postgres/pg_schemalog_store_test.go
@@ -17,7 +17,7 @@ import (
 	"github.com/xataio/pgstream/pkg/schemalog"
 )
 
-func TestStore_Fetch(t *testing.T) {
+func TestStore_FetchLast(t *testing.T) {
 	t.Parallel()
 
 	testSchema := "test_schema"
@@ -88,7 +88,69 @@ func TestStore_Fetch(t *testing.T) {
 
 			s := NewStoreWithQuerier(tc.querier)
 
-			logEntry, err := s.Fetch(context.Background(), testSchema, tc.ackedOnly)
+			logEntry, err := s.FetchLast(context.Background(), testSchema, tc.ackedOnly)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantLogEntry, logEntry)
+		})
+	}
+}
+
+func TestStore_Fetch(t *testing.T) {
+	t.Parallel()
+
+	testSchema := "test_schema"
+	testID := xid.New()
+	testLogEntry := &schemalog.LogEntry{
+		ID: testID,
+	}
+	testVersion := 1
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name    string
+		querier pglib.Querier
+
+		wantLogEntry *schemalog.LogEntry
+		wantErr      error
+	}{
+		{
+			name: "ok",
+			querier: &pgmocks.Querier{
+				QueryRowFn: func(_ context.Context, query string, args ...any) pglib.Row {
+					require.Len(t, args, 2)
+					require.Equal(t, args[0], testSchema)
+					require.Equal(t, args[1], testVersion)
+					require.Equal(t,
+						fmt.Sprintf("select id, version, schema_name, schema, created_at, acked from %q.%q where schema_name = $1 and version = $2", schemalog.SchemaName, schemalog.TableName),
+						query)
+					return &mockRow{logEntry: testLogEntry}
+				},
+			},
+
+			wantLogEntry: testLogEntry,
+			wantErr:      nil,
+		},
+		{
+			name: "error - querying rows",
+			querier: &pgmocks.Querier{
+				QueryRowFn: func(_ context.Context, query string, args ...any) pglib.Row {
+					return &mockRow{scanFn: func(...any) error { return errTest }}
+				},
+			},
+
+			wantLogEntry: nil,
+			wantErr:      errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			s := NewStoreWithQuerier(tc.querier)
+
+			logEntry, err := s.Fetch(context.Background(), testSchema, testVersion)
 			require.ErrorIs(t, err, tc.wantErr)
 			require.Equal(t, tc.wantLogEntry, logEntry)
 		})
diff --git a/pkg/schemalog/store_cache_test.go b/pkg/schemalog/store_cache_test.go
index 7510b67..afc6bef 100644
--- a/pkg/schemalog/store_cache_test.go
+++ b/pkg/schemalog/store_cache_test.go
@@ -11,7 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestStoreCache_Fetch(t *testing.T) {
+func TestStoreCache_FetchLast(t *testing.T) {
 	t.Parallel()
 
 	const testSchema = "test-schema"
@@ -34,7 +34,7 @@ func TestStoreCache_Fetch(t *testing.T) {
 		{
 			name: "ok - cache miss",
 			store: &mockStore{
-				fetchFn: func(schemaName string, ackedOnly bool) (*LogEntry, error) {
+				fetchLastFn: func(schemaName string, ackedOnly bool) (*LogEntry, error) {
 					require.Equal(t, schemaName, testSchema)
 					require.Equal(t, ackedOnly, testAckedOnly)
 					return testLogEntry, nil
@@ -48,8 +48,8 @@ func TestStoreCache_Fetch(t *testing.T) {
 		{
 			name: "ok - cache hit",
 			store: &mockStore{
-				fetchFn: func(schemaName string, ackedOnly bool) (*LogEntry, error) {
-					return nil, errors.New("fetchFn: should not be called")
+				fetchLastFn: func(schemaName string, ackedOnly bool) (*LogEntry, error) {
+					return nil, errors.New("fetchLastFn: should not be called")
 				},
 			},
 			cache: map[string]*LogEntry{
@@ -62,7 +62,7 @@ func TestStoreCache_Fetch(t *testing.T) {
 		{
 			name: "error - fetching schema",
 			store: &mockStore{
-				fetchFn: func(schemaName string, ackedOnly bool) (*LogEntry, error) {
+				fetchLastFn: func(schemaName string, ackedOnly bool) (*LogEntry, error) {
 					return nil, errTest
 				},
 			},
@@ -83,7 +83,7 @@ func TestStoreCache_Fetch(t *testing.T) {
 				s.cache = tc.cache
 			}
 
-			le, err := s.Fetch(context.Background(), testSchema, testAckedOnly)
+			le, err := s.FetchLast(context.Background(), testSchema, testAckedOnly)
 			require.ErrorIs(t, err, tc.wantErr)
 			require.Equal(t, le, tc.wantLogEntry)
 		})
@@ -230,17 +230,22 @@ func TestStoreCache_Insert(t *testing.T) {
 }
 
 type mockStore struct {
-	insertFn func(schemaName string) (*LogEntry, error)
-	fetchFn  func(schemaName string, ackedOnly bool) (*LogEntry, error)
-	ackFn    func(le *LogEntry) error
+	insertFn    func(schemaName string) (*LogEntry, error)
+	fetchLastFn func(schemaName string, ackedOnly bool) (*LogEntry, error)
+	fetchFn     func(schemaName string, version int) (*LogEntry, error)
+	ackFn       func(le *LogEntry) error
 }
 
 func (m *mockStore) Insert(_ context.Context, schemaName string) (*LogEntry, error) {
 	return m.insertFn(schemaName)
 }
 
-func (m *mockStore) Fetch(_ context.Context, schemaName string, ackedOnly bool) (*LogEntry, error) {
-	return m.fetchFn(schemaName, ackedOnly)
+func (m *mockStore) FetchLast(_ context.Context, schemaName string, ackedOnly bool) (*LogEntry, error) {
+	return m.fetchLastFn(schemaName, ackedOnly)
+}
+
+func (m *mockStore) Fetch(_ context.Context, schemaName string, version int) (*LogEntry, error) {
+	return m.fetchFn(schemaName, version)
 }
 
 func (m *mockStore) Ack(_ context.Context, le *LogEntry) error {
diff --git a/pkg/wal/processor/injector/wal_injector_test.go b/pkg/wal/processor/injector/wal_injector_test.go
index 0bef057..be3df35 100644
--- a/pkg/wal/processor/injector/wal_injector_test.go
+++ b/pkg/wal/processor/injector/wal_injector_test.go
@@ -102,7 +102,7 @@ func TestInjector_ProcessWALEvent(t *testing.T) {
 			name:  "ok - data event",
 			event: newTestDataEvent("I"),
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return testLogEntry, nil
 				},
@@ -120,7 +120,7 @@ func TestInjector_ProcessWALEvent(t *testing.T) {
 			name:  "ok - fail to inject data event",
 			event: newTestDataEvent("I"),
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					return nil, errTest
 				},
 			},
@@ -137,7 +137,7 @@ func TestInjector_ProcessWALEvent(t *testing.T) {
 			name:  "ok - fail to inject data event with invalid data",
 			event: newTestDataEvent("I"),
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					return testLogEntry, nil
 				},
 			},
@@ -164,7 +164,7 @@ func TestInjector_ProcessWALEvent(t *testing.T) {
 			name:  "error - processing event",
 			event: newTestDataEvent("I"),
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					return testLogEntry, nil
 				},
 			},
@@ -239,7 +239,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "ok - default primary key finder",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
@@ -254,7 +254,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "ok - custom id finder",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
@@ -269,7 +269,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "ok - no version provided",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
@@ -287,7 +287,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "ok - version finder provided with use LSN error",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
@@ -306,7 +306,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "error - fetching schema log entry",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					return nil, errTest
 				},
 			},
@@ -318,7 +318,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "error - schema not found",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					return nil, schemalog.ErrNoRows
 				},
 			},
@@ -330,7 +330,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "error - table not found",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
@@ -351,7 +351,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "error - filling metadata, id not found",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
@@ -373,7 +373,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "error - filling metadata, version not found",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
@@ -396,7 +396,7 @@ func TestInjector_inject(t *testing.T) {
 		{
 			name: "error - injecting column ids",
 			store: &schemalogmocks.Store{
-				FetchFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
+				FetchLastFn: func(ctx context.Context, schemaName string, ackedOnly bool) (*schemalog.LogEntry, error) {
 					require.Equal(t, testSchemaName, schemaName)
 					return newTestLogEntry(), nil
 				},
diff --git a/pkg/wal/processor/postgres/helper_test.go b/pkg/wal/processor/postgres/helper_test.go
index 12332d6..6d7fb3a 100644
--- a/pkg/wal/processor/postgres/helper_test.go
+++ b/pkg/wal/processor/postgres/helper_test.go
@@ -2,12 +2,16 @@
 
 package postgres
 
-import "github.com/xataio/pgstream/pkg/wal"
+import (
+	"context"
+
+	"github.com/xataio/pgstream/pkg/wal"
+)
 
 type mockAdapter struct {
-	walEventToQueryFn func(*wal.Event) (*query, error)
+	walEventToQueriesFn func(*wal.Event) ([]*query, error)
 }
 
-func (m *mockAdapter) walEventToQuery(e *wal.Event) (*query, error) {
-	return m.walEventToQueryFn(e)
+func (m *mockAdapter) walEventToQueries(_ context.Context, e *wal.Event) ([]*query, error) {
+	return m.walEventToQueriesFn(e)
 }
diff --git a/pkg/wal/processor/postgres/postgres_batch_writer_test.go b/pkg/wal/processor/postgres/postgres_batch_writer_test.go
index d22e106..8008b57 100644
--- a/pkg/wal/processor/postgres/postgres_batch_writer_test.go
+++ b/pkg/wal/processor/postgres/postgres_batch_writer_test.go
@@ -61,9 +61,9 @@ func TestBatchWriter_ProcessWALEvent(t *testing.T) {
 			walEvent:    testWalEvent,
 			batchSender: batchmocks.NewBatchSender[*query](),
 			adapter: &mockAdapter{
-				walEventToQueryFn: func(e *wal.Event) (*query, error) {
+				walEventToQueriesFn: func(e *wal.Event) ([]*query, error) {
 					require.Equal(t, e, testWalEvent)
-					return testInsertQuery, nil
+					return []*query{testInsertQuery}, nil
 				},
 			},
 
@@ -77,7 +77,7 @@ func TestBatchWriter_ProcessWALEvent(t *testing.T) {
 			walEvent:    testWalEvent,
 			batchSender: batchmocks.NewBatchSender[*query](),
 			adapter: &mockAdapter{
-				walEventToQueryFn: func(e *wal.Event) (*query, error) {
+				walEventToQueriesFn: func(e *wal.Event) ([]*query, error) {
 					return nil, errTest
 				},
 			},
@@ -94,9 +94,9 @@ func TestBatchWriter_ProcessWALEvent(t *testing.T) {
 				return s
 			}(),
 			adapter: &mockAdapter{
-				walEventToQueryFn: func(e *wal.Event) (*query, error) {
+				walEventToQueriesFn: func(e *wal.Event) ([]*query, error) {
 					require.Equal(t, e, testWalEvent)
-					return testInsertQuery, nil
+					return []*query{testInsertQuery}, nil
 				},
 			},
 
@@ -108,7 +108,7 @@ func TestBatchWriter_ProcessWALEvent(t *testing.T) {
 			walEvent:    testWalEvent,
 			batchSender: batchmocks.NewBatchSender[*query](),
 			adapter: &mockAdapter{
-				walEventToQueryFn: func(e *wal.Event) (*query, error) {
+				walEventToQueriesFn: func(e *wal.Event) ([]*query, error) {
 					panic(errTest)
 				},
 			},
@@ -256,11 +256,11 @@ func TestBatchWriter(t *testing.T) {
 	}
 
 	adapter := &mockAdapter{
-		walEventToQueryFn: func(e *wal.Event) (*query, error) {
-			return &query{
+		walEventToQueriesFn: func(e *wal.Event) ([]*query, error) {
+			return []*query{{
 				sql:  "INSERT INTO test(id, name) VALUES($1, $2) ON CONFLICT (id) DO NOTHING",
 				args: []any{1, "alice"},
-			}, nil
+			}}, nil
 		},
 	}
 
diff --git a/pkg/wal/processor/postgres/postgres_wal_ddl_adapter_test.go b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter_test.go
new file mode 100644
index 0000000..43bc1c5
--- /dev/null
+++ b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter_test.go
@@ -0,0 +1,408 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/rs/xid"
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/pkg/schemalog"
+	schemalogmocks "github.com/xataio/pgstream/pkg/schemalog/mocks"
+	"github.com/xataio/pgstream/pkg/wal"
+	"github.com/xataio/pgstream/pkg/wal/processor"
+)
+
+func TestDDLAdapter_walDataToQueries(t *testing.T) {
+	t.Parallel()
+
+	now := time.Now().UTC().Round(time.Second)
+	nowStr := now.Format("2006-01-02 15:04:05")
+	id := xid.New()
+	testWalData := &wal.Data{
+		Action: "I",
+		Schema: schemalog.SchemaName,
+		Table:  schemalog.TableName,
+		Columns: []wal.Column{
+			{ID: "id", Name: "id", Type: "text", Value: id.String()},
+			{ID: "version", Name: "version", Type: "integer", Value: 1},
+			{ID: "schema_name", Name: "schema_name", Type: "text", Value: testSchema},
+			{ID: "created_at", Name: "created_at", Type: "timestamp", Value: nowStr},
+		},
+	}
+
+	table1 := "table1"
+	table2 := "table2"
+
+	testLogEntry := &schemalog.LogEntry{
+		ID:         id,
+		Version:    0,
+		SchemaName: testSchema,
+		CreatedAt:  schemalog.NewSchemaCreatedAtTimestamp(now),
+	}
+
+	tests := []struct {
+		name            string
+		querier         schemalogQuerier
+		schemaDiffer    schemaDiffer
+		logEntryAdapter logEntryAdapter
+
+		wantQueries []*query
+		wantErr     error
+	}{
+		{
+			name: "ok",
+			querier: &schemalogmocks.Store{
+				FetchFn: func(ctx context.Context, schemaName string, version int) (*schemalog.LogEntry, error) {
+					require.Equal(t, testSchema, schemaName)
+					require.Equal(t, int(0), version)
+					return testLogEntry, nil
+				},
+			},
+			schemaDiffer: func(old, new *schemalog.LogEntry) *schemalog.Diff {
+				return &schemalog.Diff{
+					TablesChanged: []schemalog.TableDiff{
+						{
+							TableName: table2,
+							TableNameChange: &schemalog.ValueChange[string]{
+								Old: table1,
+								New: table2,
+							},
+						},
+					},
+				}
+			},
+			logEntryAdapter: processor.WalDataToLogEntry,
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s RENAME TO %s", quotedTableName(testSchema, table1), table2),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name:            "error - converting event to log entry",
+			logEntryAdapter: func(d *wal.Data) (*schemalog.LogEntry, error) { return nil, errTest },
+
+			wantQueries: nil,
+			wantErr:     errTest,
+		},
+		{
+			name: "error - fetching schema log",
+			querier: &schemalogmocks.Store{
+				FetchFn: func(ctx context.Context, schemaName string, version int) (*schemalog.LogEntry, error) {
+					return nil, errTest
+				},
+			},
+			logEntryAdapter: processor.WalDataToLogEntry,
+
+			wantQueries: nil,
+			wantErr:     errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			a := ddlAdapter{
+				schemalogQuerier: tc.querier,
+				schemaDiffer:     tc.schemaDiffer,
+				logEntryAdapter:  tc.logEntryAdapter,
+			}
+
+			queries, err := a.walDataToQueries(context.Background(), testWalData)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantQueries, queries)
+		})
+	}
+}
+
+func TestDDLAdapter_schemaDiffToQueries(t *testing.T) {
+	t.Parallel()
+
+	table1 := "test-table-1"
+	table2 := "test-table-2"
+	defaultAge := "0"
+
+	tests := []struct {
+		name string
+		diff *schemalog.Diff
+
+		wantQueries []*query
+		wantErr     error
+	}{
+		{
+			name: "ok - empty schema diff",
+			diff: &schemalog.Diff{},
+
+			wantQueries: []*query{},
+			wantErr:     nil,
+		},
+		{
+			name: "ok - table removed",
+			diff: &schemalog.Diff{
+				TablesRemoved: []schemalog.Table{
+					{Name: table1},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("DROP TABLE IF EXISTS %s", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table added",
+			diff: &schemalog.Diff{
+				TablesAdded: []schemalog.Table{
+					{
+						Name:              table1,
+						PrimaryKeyColumns: []string{"id"},
+						Columns: []schemalog.Column{
+							{Name: "id", DataType: "uuid", Nullable: false, Unique: true},
+							{Name: "name", DataType: "text", Nullable: true, Unique: true},
+							{Name: "age", DataType: "int", Nullable: false, DefaultValue: &defaultAge},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s ( id uuid NOT NULL,\n name text,\n age int NOT NULL DEFAULT 0,\n UNIQUE (name),\n PRIMARY KEY (id)\n)", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table renamed",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table2,
+						TableNameChange: &schemalog.ValueChange[string]{
+							Old: table1,
+							New: table2,
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s RENAME TO %s", quotedTableName(testSchema, table1), table2),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column dropped",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsRemoved: []schemalog.Column{
+							{Name: "age"},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s DROP COLUMN age", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column added",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsAdded: []schemalog.Column{
+							{Name: "age", DataType: "int", Nullable: false, DefaultValue: &defaultAge},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s ADD COLUMN age int NOT NULL DEFAULT 0", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column renamed",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsChanged: []schemalog.ColumnDiff{
+							{
+								ColumnName: "name",
+								NameChange: &schemalog.ValueChange[string]{Old: "name", New: "new_name"},
+							},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s RENAME COLUMN name TO new_name", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column type changed",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsChanged: []schemalog.ColumnDiff{
+							{
+								ColumnName: "name",
+								TypeChange: &schemalog.ValueChange[string]{Old: "uuid", New: "int"},
+							},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s ALTER COLUMN name TYPE int", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column changed to nullable",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsChanged: []schemalog.ColumnDiff{
+							{
+								ColumnName: "name",
+								NullChange: &schemalog.ValueChange[bool]{Old: false, New: true},
+							},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s ALTER COLUMN name DROP NOT NULL", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column changed to not nullable",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsChanged: []schemalog.ColumnDiff{
+							{
+								ColumnName: "name",
+								NullChange: &schemalog.ValueChange[bool]{Old: true, New: false},
+							},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s ALTER COLUMN name SET NOT NULL", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column default removed",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsChanged: []schemalog.ColumnDiff{
+							{
+								ColumnName:    "age",
+								DefaultChange: &schemalog.ValueChange[*string]{Old: &defaultAge, New: nil},
+							},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s ALTER COLUMN age DROP DEFAULT", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table changed, column default changed",
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						TableName: table1,
+						ColumnsChanged: []schemalog.ColumnDiff{
+							{
+								ColumnName:    "age",
+								DefaultChange: &schemalog.ValueChange[*string]{Old: nil, New: &defaultAge},
+							},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("ALTER TABLE %s ALTER COLUMN age SET DEFAULT 0", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			ddlAdapter := newDDLAdapter(nil)
+
+			queries, err := ddlAdapter.schemaDiffToQueries(testSchema, tc.diff)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantQueries, queries)
+		})
+	}
+}
diff --git a/pkg/wal/processor/postgres/postgres_wal_adapter_test.go b/pkg/wal/processor/postgres/postgres_wal_dml_adapter_test.go
similarity index 67%
rename from pkg/wal/processor/postgres/postgres_wal_adapter_test.go
rename to pkg/wal/processor/postgres/postgres_wal_dml_adapter_test.go
index 4a3289d..521c374 100644
--- a/pkg/wal/processor/postgres/postgres_wal_adapter_test.go
+++ b/pkg/wal/processor/postgres/postgres_wal_dml_adapter_test.go
@@ -8,92 +8,10 @@ import (
 
 	"github.com/rs/xid"
 	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal"
 )
 
-func TestAdapter_walEventToQuery(t *testing.T) {
-	t.Parallel()
-
-	testCommitPosition := wal.CommitPosition("1/0")
-	testTable := "table"
-	testSchema := "test"
-
-	tests := []struct {
-		name  string
-		event *wal.Event
-
-		wantQuery *query
-		wantErr   error
-	}{
-		{
-			name: "ok - no data",
-			event: &wal.Event{
-				CommitPosition: testCommitPosition,
-			},
-
-			wantQuery: &query{},
-			wantErr:   nil,
-		},
-		{
-			name: "ok - schema event",
-			event: &wal.Event{
-				Data: &wal.Data{
-					Schema: schemalog.SchemaName,
-					Table:  schemalog.TableName,
-				},
-				CommitPosition: testCommitPosition,
-			},
-
-			wantQuery: &query{},
-			wantErr:   nil,
-		},
-		{
-			name: "ok - supported data event",
-			event: &wal.Event{
-				Data: &wal.Data{
-					Action: "T",
-					Schema: testSchema,
-					Table:  testTable,
-				},
-				CommitPosition: testCommitPosition,
-			},
-
-			wantQuery: &query{
-				sql: fmt.Sprintf("TRUNCATE %s", quotedTableName(testSchema, testTable)),
-			},
-			wantErr: nil,
-		},
-		{
-			name: "ok - unsupported data event",
-			event: &wal.Event{
-				Data: &wal.Data{
-					Action: "X",
-					Schema: testSchema,
-					Table:  testTable,
-				},
-				CommitPosition: testCommitPosition,
-			},
-
-			wantQuery: &query{},
-			wantErr:   nil,
-		},
-	}
-
-	for _, tc := range tests {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			a := &adapter{}
-			query, err := a.walEventToQuery(tc.event)
-			require.ErrorIs(t, err, tc.wantErr)
-			require.Equal(t, tc.wantQuery, query)
-		})
-	}
-}
-
-func TestAdapter_walDataToQuery(t *testing.T) {
+func TestDMLAdapter_walDataToQuery(t *testing.T) {
 	t.Parallel()
 
 	testTableID := xid.New()
@@ -221,7 +139,7 @@ func TestAdapter_walDataToQuery(t *testing.T) {
 				},
 			},
 
-			wantQuery: nil,
+			wantQuery: &query{},
 		},
 	}
 
@@ -229,7 +147,7 @@ func TestAdapter_walDataToQuery(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			a := &adapter{}
+			a := &dmlAdapter{}
 			query := a.walDataToQuery(tc.walData)
 			require.Equal(t, tc.wantQuery, query)
 		})