Repository: xataio/pgstream
Pull request #131 (2025-02-06 10:52:07)

Add enriched schema diff computation
This PR adds a more complete computation of a schema log entry diff, and it decouples the diff search store implementation specific details.

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

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/schemalog | 55.50% (**+15.13%**) | :tada: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/schemalog/schema_diff.go | 98.25% (**+98.25%**) | 57 (+57) | 56 (+56) | 1 (+1) | :star2: |

_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/pkg/schemalog/schema_diff_test.go

</details>


Patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 16e5448..46e8ee8 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -65,7 +65,7 @@ jobs:
           # Require: The version of golangci-lint to use.
           # When `install-mode` is `binary` (default) the value can be v1.2 or v1.2.3 or `latest` to use the latest version.
           # When `install-mode` is `goinstall` the value can be v1.2.3, `latest`, or the hash of a commit.
-          version: v1.58.1
+          version: v1.60.1
 
           # Optional: golangci-lint command line arguments.
           #
diff --git a/pkg/schemalog/log_entry.go b/pkg/schemalog/log_entry.go
index b5b7989..b26b95f 100644
--- a/pkg/schemalog/log_entry.go
+++ b/pkg/schemalog/log_entry.go
@@ -89,13 +89,6 @@ func (m *LogEntry) After(other *LogEntry) bool {
 	return m.ID.Compare(other.ID) > 0
 }
 
-func (m *LogEntry) Diff(previous *LogEntry) *SchemaDiff {
-	if previous == nil {
-		return m.Schema.Diff(&Schema{})
-	}
-	return m.Schema.Diff(&previous.Schema)
-}
-
 func (m *LogEntry) GetTableByName(tableName string) (Table, bool) {
 	return m.Schema.getTableByName(tableName)
 }
diff --git a/pkg/schemalog/schema.go b/pkg/schemalog/schema.go
index a913ca2..cf3df33 100644
--- a/pkg/schemalog/schema.go
+++ b/pkg/schemalog/schema.go
@@ -67,35 +67,6 @@ func (s *Schema) IsEqual(other *Schema) bool {
 	}
 }
 
-func (s *Schema) Diff(previous *Schema) *SchemaDiff {
-	var d SchemaDiff
-
-	// if a table ID exists in previous, but not s: remove the table
-	for _, previousTable := range previous.Tables {
-		if !s.hasTableID(previousTable.PgstreamID) {
-			d.TablesToRemove = append(d.TablesToRemove, previousTable)
-		}
-	}
-
-	for i, table := range s.Tables {
-		if previousTable := previous.getTableByID(table.PgstreamID); previousTable != nil {
-			d.ColumnsToAdd = append(d.ColumnsToAdd, diffColumns(&s.Tables[i], previousTable)...)
-			if hasPrimaryKeyChanged(previousTable.PrimaryKeyColumns, table.PrimaryKeyColumns) {
-				d.PrimaryKeyChange = append(d.PrimaryKeyChange, table.Name)
-			}
-			if hasUniqueNotNullChanged(previousTable, &s.Tables[i]) {
-				d.UniqueNotNullChange = append(d.UniqueNotNullChange, table.Name)
-			}
-		} else {
-			// if the "old" schema does not have the table, we can add all
-			// columns without checking further.
-			d.ColumnsToAdd = append(d.ColumnsToAdd, table.Columns...)
-		}
-	}
-
-	return &d
-}
-
 func (s *Schema) getTableByName(tableName string) (Table, bool) {
 	for _, t := range s.Tables {
 		if t.Name == tableName {
@@ -105,19 +76,6 @@ func (s *Schema) getTableByName(tableName string) (Table, bool) {
 	return Table{}, false
 }
 
-func (s *Schema) getTableByID(pgstreamID string) *Table {
-	for i := range s.Tables {
-		if s.Tables[i].PgstreamID == pgstreamID {
-			return &s.Tables[i]
-		}
-	}
-	return nil
-}
-
-func (s *Schema) hasTableID(pgstreamID string) bool {
-	return s.getTableByID(pgstreamID) != nil
-}
-
 func (t *Table) IsEqual(other *Table) bool {
 	switch {
 	case t == nil && other == nil:
@@ -187,17 +145,6 @@ func (c *Column) IsEqual(other *Column) bool {
 	}
 }
 
-type SchemaDiff struct {
-	TablesToRemove      []Table
-	ColumnsToAdd        []Column
-	PrimaryKeyChange    []string
-	UniqueNotNullChange []string
-}
-
-func (d *SchemaDiff) Empty() bool {
-	return len(d.TablesToRemove) == 0 && len(d.ColumnsToAdd) == 0
-}
-
 func unorderedColumnsEqual(a, b []Column) bool {
 	if len(a) != len(b) {
 		return false
@@ -220,40 +167,3 @@ func unorderedColumnsEqual(a, b []Column) bool {
 
 	return true
 }
-
-func diffColumns(new, old *Table) []Column {
-	var colsAdded []Column
-
-	for _, newCol := range new.Columns {
-		var found bool
-
-		for _, oldCol := range old.Columns {
-			if newCol.PgstreamID == oldCol.PgstreamID {
-				found = true
-				break
-			}
-		}
-
-		if !found {
-			colsAdded = append(colsAdded, newCol)
-		}
-	}
-
-	return colsAdded
-}
-
-func hasPrimaryKeyChanged(old, new []string) bool {
-	slices.Sort(old)
-	slices.Sort(new)
-	return !slices.Equal(old, new)
-}
-
-func hasUniqueNotNullChanged(old, new *Table) bool {
-	// we only care about unique not null changes if there's no primary key defined
-	if len(old.PrimaryKeyColumns) != 0 || len(new.PrimaryKeyColumns) != 0 {
-		return false
-	}
-	oldUniqueNotNullCol := old.GetFirstUniqueNotNullColumn()
-	newUniqueNotNullCol := new.GetFirstUniqueNotNullColumn()
-	return !oldUniqueNotNullCol.IsEqual(newUniqueNotNullCol)
-}
diff --git a/pkg/schemalog/schema_diff.go b/pkg/schemalog/schema_diff.go
new file mode 100644
index 0000000..a83626c
--- /dev/null
+++ b/pkg/schemalog/schema_diff.go
@@ -0,0 +1,168 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package schemalog
+
+import (
+	"slices"
+)
+
+type Diff struct {
+	TablesRemoved []Table
+	TablesAdded   []Table
+	TablesChanged []TableDiff
+}
+
+type TableDiff struct {
+	TableName             string
+	TablePgstreamID       string
+	TableNameChange       *ValueChange[string]
+	TablePrimaryKeyChange *ValueChange[[]string]
+	ColumnsAdded          []Column
+	ColumnsRemoved        []Column
+	ColumnsChanged        []ColumnDiff
+}
+
+type ColumnDiff struct {
+	ColumnName       string
+	ColumnPgstreamID string
+	NameChange       *ValueChange[string]
+	TypeChange       *ValueChange[string]
+	UniqueChange     *ValueChange[bool]
+	NullChange       *ValueChange[bool]
+	DefaultChange    *ValueChange[*string]
+}
+
+type ValueChange[T any] struct {
+	Old, New T
+}
+
+func (d *Diff) IsEmpty() bool {
+	return len(d.TablesAdded) == 0 && len(d.TablesChanged) == 0 && len(d.TablesRemoved) == 0
+}
+
+func (td *TableDiff) IsEmpty() bool {
+	return len(td.ColumnsAdded) == 0 && len(td.ColumnsRemoved) == 0 && len(td.ColumnsChanged) == 0 && td.TableNameChange == nil && td.TablePrimaryKeyChange == nil
+}
+
+func (cd *ColumnDiff) IsEmpty() bool {
+	return cd.TypeChange == nil && cd.NameChange == nil && cd.DefaultChange == nil && cd.NullChange == nil && cd.UniqueChange == nil
+}
+
+func ComputeSchemaDiff(old, new *LogEntry) *Diff {
+	switch {
+	case old == nil && new == nil:
+		return &Diff{}
+	case old == nil:
+		old = &LogEntry{}
+	case new == nil:
+		new = &LogEntry{}
+	}
+
+	diff := &Diff{}
+	newTableMap := getSchemaTableMap(&new.Schema)
+	// if a table ID exists in the old schema, but not in the new, remove the table
+	for _, oldTable := range old.Schema.Tables {
+		if _, found := newTableMap[oldTable.PgstreamID]; !found {
+			diff.TablesRemoved = append(diff.TablesRemoved, oldTable)
+		}
+	}
+
+	oldTableMap := getSchemaTableMap(&old.Schema)
+	for id, newTable := range newTableMap {
+		oldTable, found := oldTableMap[id]
+		// if the table is not on the old schema, add it
+		if !found {
+			diff.TablesAdded = append(diff.TablesAdded, newTable)
+			continue
+		}
+
+		// both schemas have the table, check for changes
+		tableDiff := computeTableDiff(&oldTable, &newTable)
+		if !tableDiff.IsEmpty() {
+			diff.TablesChanged = append(diff.TablesChanged, *tableDiff)
+		}
+	}
+
+	return diff
+}
+
+func computeTableDiff(old, new *Table) *TableDiff {
+	diff := &TableDiff{
+		TableName:       new.Name,
+		TablePgstreamID: new.PgstreamID,
+	}
+
+	if old.Name != new.Name {
+		diff.TableNameChange = &ValueChange[string]{Old: old.Name, New: new.Name}
+	}
+
+	if !slices.Equal(old.PrimaryKeyColumns, new.PrimaryKeyColumns) {
+		diff.TablePrimaryKeyChange = &ValueChange[[]string]{Old: old.PrimaryKeyColumns, New: new.PrimaryKeyColumns}
+	}
+
+	newColumnMap := getTableColumnMap(new)
+	// if a column ID exists in the old table, but not in the new, remove the column
+	for _, col := range old.Columns {
+		if _, found := newColumnMap[col.PgstreamID]; !found {
+			diff.ColumnsRemoved = append(diff.ColumnsRemoved, col)
+		}
+	}
+
+	oldColumnMap := getTableColumnMap(old)
+	for id, newCol := range newColumnMap {
+		oldCol, found := oldColumnMap[id]
+		if !found {
+			diff.ColumnsAdded = append(diff.ColumnsAdded, newCol)
+			continue
+		}
+
+		// both tables have the column, check for changes
+		colDiff := computeColumnDiff(&oldCol, &newCol)
+		if !colDiff.IsEmpty() {
+			diff.ColumnsChanged = append(diff.ColumnsChanged, *colDiff)
+		}
+	}
+
+	return diff
+}
+
+func computeColumnDiff(old, new *Column) *ColumnDiff {
+	diff := &ColumnDiff{
+		ColumnName:       new.Name,
+		ColumnPgstreamID: new.PgstreamID,
+	}
+
+	if old.DataType != new.DataType {
+		diff.TypeChange = &ValueChange[string]{Old: old.DataType, New: new.DataType}
+	}
+	if old.Name != new.Name {
+		diff.NameChange = &ValueChange[string]{Old: old.Name, New: new.Name}
+	}
+	if old.DefaultValue != new.DefaultValue {
+		diff.DefaultChange = &ValueChange[*string]{Old: old.DefaultValue, New: new.DefaultValue}
+	}
+	if old.Unique != new.Unique {
+		diff.UniqueChange = &ValueChange[bool]{Old: old.Unique, New: new.Unique}
+	}
+	if old.Nullable != new.Nullable {
+		diff.NullChange = &ValueChange[bool]{Old: old.Nullable, New: new.Nullable}
+	}
+
+	return diff
+}
+
+func getSchemaTableMap(s *Schema) map[string]Table {
+	tableMap := make(map[string]Table, len(s.Tables))
+	for _, t := range s.Tables {
+		tableMap[t.PgstreamID] = t
+	}
+	return tableMap
+}
+
+func getTableColumnMap(t *Table) map[string]Column {
+	columnMap := make(map[string]Column, len(t.Columns))
+	for _, c := range t.Columns {
+		columnMap[c.PgstreamID] = c
+	}
+	return columnMap
+}
diff --git a/pkg/wal/processor/search/store/search_store.go b/pkg/wal/processor/search/store/search_store.go
index 8021771..5e3ff09 100644
--- a/pkg/wal/processor/search/store/search_store.go
+++ b/pkg/wal/processor/search/store/search_store.go
@@ -142,18 +142,18 @@ func (s *Store) ApplySchemaChange(ctx context.Context, newEntry *schemalog.LogEn
 		}
 	}
 
-	changes := newEntry.Diff(existingLogEntry)
+	changes := schemalog.ComputeSchemaDiff(existingLogEntry, newEntry)
+
 	// for now only log a warning when a reindexing is required, since the data
 	// can still be indexed, old data will need a reindex to take into account
 	// the new identity for search to be effective
 	//
 	// TODO: in the future, we will trigger a reindex automatically in this
 	// situation
-	for _, tbl := range changes.PrimaryKeyChange {
-		s.logger.Warn(nil, fmt.Sprintf("primary key identity column changed for table %s, reindexing required", tbl))
-	}
-	for _, tbl := range changes.UniqueNotNullChange {
-		s.logger.Warn(nil, fmt.Sprintf("unique not null identity column changed for table %s, reindexing required", tbl))
+	for _, tbl := range changes.TablesChanged {
+		if tbl.TablePrimaryKeyChange != nil {
+			s.logger.Warn(nil, fmt.Sprintf("primary key identity column changed for table %s, reindexing required", tbl.TableName))
+		}
 	}
 
 	if err := s.updateMapping(ctx, newEntry.SchemaName, newEntry, changes); err != nil {
@@ -306,16 +306,16 @@ func (s *Store) createSchema(ctx context.Context, schemaName string) error {
 	return nil
 }
 
-func (s *Store) updateMapping(ctx context.Context, schemaName string, logEntry *schemalog.LogEntry, diff *schemalog.SchemaDiff) error {
+func (s *Store) updateMapping(ctx context.Context, schemaName string, logEntry *schemalog.LogEntry, diff *schemalog.Diff) error {
 	index := s.indexNameAdapter.SchemaNameToIndex(schemaName)
-	if diff != nil {
-		if err := s.updateMappingAddNewColumns(ctx, index, diff.ColumnsToAdd); err != nil {
+	if diff != nil && !diff.IsEmpty() {
+		if err := s.updateMappingAddNewColumns(ctx, index, s.getAllNewColumns(diff)); err != nil {
 			return fmt.Errorf("failed to add new columns: %w", mapError(err))
 		}
 
-		if len(diff.TablesToRemove) > 0 {
-			tableIDs := make([]string, 0, len(diff.TablesToRemove))
-			for _, table := range diff.TablesToRemove {
+		if len(diff.TablesRemoved) > 0 {
+			tableIDs := make([]string, 0, len(diff.TablesRemoved))
+			for _, table := range diff.TablesRemoved {
 				tableIDs = append(tableIDs, table.PgstreamID)
 			}
 			if err := s.deleteTableDocuments(ctx, index, tableIDs); err != nil {
@@ -459,7 +459,8 @@ func (s *Store) ensureSchemaMapping(ctx context.Context, schemaName string, meta
 		// if the schema didn't exist, but there's a log entry in the schemalog,
 		// we need to reset it to the latest known mapping
 		if metadata != nil && !metadata.IsEmpty() {
-			if err := s.updateMapping(ctx, schemaName, metadata, metadata.Diff(&schemalog.LogEntry{})); err != nil {
+			changes := schemalog.ComputeSchemaDiff(&schemalog.LogEntry{}, metadata)
+			if err := s.updateMapping(ctx, schemaName, metadata, changes); err != nil {
 				return fmt.Errorf("updating mapping for missing schema: %w", err)
 			}
 		}
@@ -467,6 +468,18 @@ func (s *Store) ensureSchemaMapping(ctx context.Context, schemaName string, meta
 	return nil
 }
 
+func (s *Store) getAllNewColumns(diff *schemalog.Diff) []schemalog.Column {
+	cols := []schemalog.Column{}
+	for _, tbl := range diff.TablesAdded {
+		cols = append(cols, tbl.Columns...)
+	}
+
+	for _, tbl := range diff.TablesChanged {
+		cols = append(cols, tbl.ColumnsAdded...)
+	}
+	return cols
+}
+
 func mapError(err error) error {
 	if errors.As(err, &searchstore.RetryableError{}) {
 		return fmt.Errorf("%w: %w", search.ErrRetriable, err)


Test patch:
diff --git a/pkg/schemalog/log_entry_test.go b/pkg/schemalog/log_entry_test.go
deleted file mode 100644
index 937a15f..0000000
--- a/pkg/schemalog/log_entry_test.go
+++ /dev/null
@@ -1,221 +0,0 @@
-// SPDX-License-Identifier: Apache-2.0
-
-package schemalog
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/require"
-)
-
-func TestLogEntry_Diff(t *testing.T) {
-	nopDiff := &SchemaDiff{}
-
-	tcs := map[string]struct {
-		old  *LogEntry
-		new  *LogEntry
-		want *SchemaDiff
-	}{
-		"table removed": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{PgstreamID: "1"},
-					},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{},
-				},
-			},
-			want: &SchemaDiff{
-				TablesToRemove: []Table{
-					{PgstreamID: "1"},
-				},
-			},
-		},
-		"table added without columns": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{PgstreamID: "1"},
-					},
-				},
-			},
-			want: nopDiff,
-		},
-		"table renamed": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{PgstreamID: "1", Name: "old"},
-					},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{PgstreamID: "1", Name: "new"},
-					},
-				},
-			},
-			want: nopDiff,
-		},
-		"new table added with columns": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns: []Column{
-								{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-								{Name: "col2", PgstreamID: "1-2", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-							},
-						},
-					},
-				},
-			},
-			want: &SchemaDiff{
-				ColumnsToAdd: []Column{
-					{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-					{Name: "col2", PgstreamID: "1-2", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-				},
-			},
-		},
-		"column added": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns: []Column{
-								{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-							},
-						},
-					},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns: []Column{
-								{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-								{Name: "col2", PgstreamID: "1-2", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-							},
-						},
-					},
-				},
-			},
-			want: &SchemaDiff{
-				ColumnsToAdd: []Column{
-					{Name: "col2", PgstreamID: "1-2", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-				},
-			},
-		},
-		"column removed": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns: []Column{
-								{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-							},
-						},
-					},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns:    []Column{},
-						},
-					},
-				},
-			},
-			want: nopDiff,
-		},
-		"table and columns removed": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns: []Column{
-								{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-							},
-						},
-					},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{},
-				},
-			},
-			want: &SchemaDiff{
-				TablesToRemove: []Table{
-					{
-						PgstreamID: "1",
-						Columns: []Column{
-							{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-						},
-					},
-				},
-			},
-		},
-		"column renamed": {
-			old: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns: []Column{
-								{Name: "col1", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-							},
-						},
-					},
-				},
-			},
-			new: &LogEntry{
-				Schema: Schema{
-					Tables: []Table{
-						{
-							PgstreamID: "1",
-							Columns: []Column{
-								{Name: "col1-renamed", PgstreamID: "1-1", DataType: "text", Nullable: true, DefaultValue: ptr("a")},
-							},
-						},
-					},
-				},
-			},
-			want: nopDiff,
-		},
-	}
-
-	for name, tc := range tcs {
-		t.Run(name, func(t *testing.T) {
-			got := tc.new.Diff(tc.old)
-			require.Equal(t, tc.want, got)
-		})
-	}
-}
-
-func ptr[T any](in T) *T {
-	return &in
-}
diff --git a/pkg/schemalog/schema_diff_test.go b/pkg/schemalog/schema_diff_test.go
new file mode 100644
index 0000000..8a83592
--- /dev/null
+++ b/pkg/schemalog/schema_diff_test.go
@@ -0,0 +1,417 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package schemalog
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	table1 = "table-1"
+	table2 = "table-2"
+	col1   = "col-1"
+	col2   = "col-2"
+	id1    = "1"
+	id2    = "2"
+)
+
+func Test_ComputeSchemaDiff(t *testing.T) {
+	t.Parallel()
+
+	testTable := func(name, id string, cols ...[]Column) Table {
+		tableCols := []Column{
+			{PgstreamID: id + "_1", Name: "col-1"},
+		}
+		if len(cols) > 0 {
+			tableCols = cols[0]
+		}
+		return Table{
+			PgstreamID:        id,
+			Name:              name,
+			Columns:           tableCols,
+			PrimaryKeyColumns: []string{"col-1"},
+		}
+	}
+
+	tests := []struct {
+		name      string
+		newSchema *LogEntry
+		oldSchema *LogEntry
+
+		wantDiff *Diff
+	}{
+		{
+			name: "no diff",
+			newSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1)},
+				},
+			},
+			oldSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1)},
+				},
+			},
+
+			wantDiff: &Diff{},
+		},
+		{
+			name:      "no diff - no schemas",
+			newSchema: nil,
+			oldSchema: nil,
+
+			wantDiff: &Diff{},
+		},
+		{
+			name: "no old schema",
+			newSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1)},
+				},
+			},
+			oldSchema: nil,
+
+			wantDiff: &Diff{
+				TablesAdded: []Table{testTable(table1, id1)},
+			},
+		},
+		{
+			name:      "no new schema",
+			newSchema: nil,
+			oldSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1)},
+				},
+			},
+
+			wantDiff: &Diff{
+				TablesRemoved: []Table{testTable(table1, id1)},
+			},
+		},
+		{
+			name: "table added and removed",
+			newSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table2, id2)},
+				},
+			},
+			oldSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1)},
+				},
+			},
+
+			wantDiff: &Diff{
+				TablesRemoved: []Table{testTable(table1, id1)},
+				TablesAdded:   []Table{testTable(table2, id2)},
+			},
+		},
+		{
+			name: "table name changed",
+			newSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table2, id1)},
+				},
+			},
+			oldSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1)},
+				},
+			},
+
+			wantDiff: &Diff{
+				TablesChanged: []TableDiff{
+					{
+						TableName:       table2,
+						TablePgstreamID: id1,
+						TableNameChange: &ValueChange[string]{Old: table1, New: table2},
+					},
+				},
+			},
+		},
+		{
+			name: "table primary key changed",
+			newSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{
+						{
+							PgstreamID: id1,
+							Name:       table1,
+							Columns: []Column{
+								{PgstreamID: id1 + "_1", Name: "col-1"},
+								{PgstreamID: id1 + "_2", Name: "col-2"},
+							},
+							PrimaryKeyColumns: []string{"col-2"},
+						},
+					},
+				},
+			},
+			oldSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{
+						{
+							PgstreamID: id1,
+							Name:       table1,
+							Columns: []Column{
+								{PgstreamID: id1 + "_1", Name: "col-1"},
+								{PgstreamID: id1 + "_2", Name: "col-2"},
+							},
+							PrimaryKeyColumns: []string{"col-1"},
+						},
+					},
+				},
+			},
+
+			wantDiff: &Diff{
+				TablesChanged: []TableDiff{
+					{
+						TableName:             table1,
+						TablePgstreamID:       id1,
+						TablePrimaryKeyChange: &ValueChange[[]string]{Old: []string{"col-1"}, New: []string{"col-2"}},
+					},
+				},
+			},
+		},
+		{
+			name: "columns added and removed",
+			newSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1, []Column{
+						{
+							Name:       col2,
+							PgstreamID: id2,
+						},
+					})},
+				},
+			},
+			oldSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1, []Column{
+						{
+							Name:       col1,
+							PgstreamID: id1,
+						},
+					})},
+				},
+			},
+
+			wantDiff: &Diff{
+				TablesChanged: []TableDiff{
+					{
+						TableName:       table1,
+						TablePgstreamID: id1,
+						ColumnsAdded: []Column{
+							{
+								Name:       col2,
+								PgstreamID: id2,
+							},
+						},
+						ColumnsRemoved: []Column{
+							{
+								Name:       col1,
+								PgstreamID: id1,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "column name change",
+			newSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1, []Column{
+						{
+							Name:       col2,
+							PgstreamID: id1,
+						},
+					})},
+				},
+			},
+			oldSchema: &LogEntry{
+				Schema: Schema{
+					Tables: []Table{testTable(table1, id1, []Column{
+						{
+							Name:       col1,
+							PgstreamID: id1,
+						},
+					})},
+				},
+			},
+
+			wantDiff: &Diff{
+				TablesChanged: []TableDiff{
+					{
+						TableName:       table1,
+						TablePgstreamID: id1,
+						ColumnsChanged: []ColumnDiff{
+							{
+								ColumnName:       col2,
+								ColumnPgstreamID: id1,
+								NameChange:       &ValueChange[string]{Old: col1, New: col2},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			diff := ComputeSchemaDiff(tc.oldSchema, tc.newSchema)
+			require.Equal(t, tc.wantDiff, diff)
+		})
+	}
+}
+
+func Test_computeColumnDiff(t *testing.T) {
+	t.Parallel()
+
+	default1 := "1"
+
+	tests := []struct {
+		name      string
+		oldColumn *Column
+		newColumn *Column
+
+		wantDiff *ColumnDiff
+	}{
+		{
+			name: "name change",
+			oldColumn: &Column{
+				Name:       col1,
+				PgstreamID: id1,
+			},
+			newColumn: &Column{
+				Name:       col2,
+				PgstreamID: id1,
+			},
+
+			wantDiff: &ColumnDiff{
+				ColumnName:       col2,
+				ColumnPgstreamID: id1,
+				NameChange:       &ValueChange[string]{Old: col1, New: col2},
+			},
+		},
+		{
+			name: "type change",
+			oldColumn: &Column{
+				Name:       col1,
+				PgstreamID: id1,
+				DataType:   "int",
+			},
+			newColumn: &Column{
+				Name:       col1,
+				PgstreamID: id1,
+				DataType:   "text",
+			},
+
+			wantDiff: &ColumnDiff{
+				ColumnName:       col1,
+				ColumnPgstreamID: id1,
+				TypeChange:       &ValueChange[string]{Old: "int", New: "text"},
+			},
+		},
+		{
+			name: "default change",
+			oldColumn: &Column{
+				Name:         col1,
+				PgstreamID:   id1,
+				DefaultValue: nil,
+			},
+			newColumn: &Column{
+				Name:         col1,
+				PgstreamID:   id1,
+				DefaultValue: &default1,
+			},
+
+			wantDiff: &ColumnDiff{
+				ColumnName:       col1,
+				ColumnPgstreamID: id1,
+				DefaultChange:    &ValueChange[*string]{Old: nil, New: &default1},
+			},
+		},
+		{
+			name: "unique change",
+			oldColumn: &Column{
+				Name:       col1,
+				PgstreamID: id1,
+				Unique:     true,
+			},
+			newColumn: &Column{
+				Name:       col1,
+				PgstreamID: id1,
+				Unique:     false,
+			},
+
+			wantDiff: &ColumnDiff{
+				ColumnName:       col1,
+				ColumnPgstreamID: id1,
+				UniqueChange:     &ValueChange[bool]{Old: true, New: false},
+			},
+		},
+		{
+			name: "null change",
+			oldColumn: &Column{
+				Name:       col1,
+				PgstreamID: id1,
+				Nullable:   true,
+			},
+			newColumn: &Column{
+				Name:       col1,
+				PgstreamID: id1,
+				Nullable:   false,
+			},
+
+			wantDiff: &ColumnDiff{
+				ColumnName:       col1,
+				ColumnPgstreamID: id1,
+				NullChange:       &ValueChange[bool]{Old: true, New: false},
+			},
+		},
+		{
+			name: "multiple changes",
+			oldColumn: &Column{
+				Name:         col1,
+				PgstreamID:   id1,
+				DataType:     "int",
+				DefaultValue: nil,
+				Unique:       true,
+				Nullable:     true,
+			},
+			newColumn: &Column{
+				Name:         col2,
+				PgstreamID:   id1,
+				DataType:     "text",
+				DefaultValue: &default1,
+				Unique:       false,
+				Nullable:     false,
+			},
+
+			wantDiff: &ColumnDiff{
+				ColumnName:       col2,
+				ColumnPgstreamID: id1,
+				NameChange:       &ValueChange[string]{Old: col1, New: col2},
+				TypeChange:       &ValueChange[string]{Old: "int", New: "text"},
+				DefaultChange:    &ValueChange[*string]{Old: nil, New: &default1},
+				UniqueChange:     &ValueChange[bool]{Old: true, New: false},
+				NullChange:       &ValueChange[bool]{Old: true, New: false},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			diff := computeColumnDiff(tc.oldColumn, tc.newColumn)
+			require.Equal(t, tc.wantDiff, diff)
+		})
+	}
+}
diff --git a/pkg/schemalog/schema_test.go b/pkg/schemalog/schema_test.go
index fcb4a2d..3e57ad0 100644
--- a/pkg/schemalog/schema_test.go
+++ b/pkg/schemalog/schema_test.go
@@ -8,142 +8,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestSchema_Diff(t *testing.T) {
-	t.Parallel()
-
-	testTableName := "test-table"
-	testTable := func() Table {
-		return Table{
-			PgstreamID: "1",
-			Name:       testTableName,
-			Columns: []Column{
-				{PgstreamID: "1_1", Name: "col-1"},
-			},
-			PrimaryKeyColumns: []string{"col-1"},
-		}
-	}
-
-	tests := []struct {
-		name      string
-		schema    Schema
-		oldSchema Schema
-
-		wantDiff *SchemaDiff
-	}{
-		{
-			name: "columns to add",
-			schema: Schema{
-				Tables: []Table{testTable()},
-			},
-			oldSchema: Schema{
-				Tables: []Table{
-					{
-						PgstreamID:        "1",
-						Name:              testTableName,
-						PrimaryKeyColumns: []string{"col-1"},
-					},
-				},
-			},
-
-			wantDiff: &SchemaDiff{
-				ColumnsToAdd: []Column{
-					{PgstreamID: "1_1", Name: "col-1"},
-				},
-			},
-		},
-		{
-			name: "columns to add with non existing table",
-			schema: Schema{
-				Tables: []Table{testTable()},
-			},
-			oldSchema: Schema{
-				Tables: []Table{},
-			},
-
-			wantDiff: &SchemaDiff{
-				ColumnsToAdd: []Column{
-					{PgstreamID: "1_1", Name: "col-1"},
-				},
-			},
-		},
-		{
-			name: "tables to remove",
-			schema: Schema{
-				Tables: []Table{},
-			},
-			oldSchema: Schema{
-				Tables: []Table{testTable()},
-			},
-
-			wantDiff: &SchemaDiff{
-				TablesToRemove: []Table{testTable()},
-			},
-		},
-		{
-			name: "primary key changed",
-			schema: Schema{
-				Tables: []Table{testTable()},
-			},
-			oldSchema: Schema{
-				Tables: []Table{
-					{
-						PgstreamID: "1",
-						Name:       testTableName,
-						Columns: []Column{
-							{PgstreamID: "1_1", Name: "col-1"},
-						},
-					},
-				},
-			},
-
-			wantDiff: &SchemaDiff{
-				PrimaryKeyChange: []string{testTableName},
-			},
-		},
-		{
-			name: "unique not null changed",
-			schema: Schema{
-				Tables: []Table{
-					{
-						PgstreamID: "1",
-						Name:       testTableName,
-						Columns: []Column{
-							{PgstreamID: "1_1", Name: "col-1", Unique: true, Nullable: false},
-							{PgstreamID: "1_2", Name: "col-2", Unique: true, Nullable: false},
-						},
-					},
-				},
-			},
-			oldSchema: Schema{
-				Tables: []Table{
-					{
-						PgstreamID: "1",
-						Name:       testTableName,
-						Columns: []Column{
-							{PgstreamID: "1_1", Name: "col-1", Unique: false, Nullable: true},
-							{PgstreamID: "1_2", Name: "col-2", Unique: true, Nullable: false},
-						},
-					},
-				},
-			},
-
-			wantDiff: &SchemaDiff{
-				UniqueNotNullChange: []string{testTableName},
-			},
-		},
-	}
-
-	for _, tc := range tests {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			diff := tc.schema.Diff(&tc.oldSchema)
-			require.Equal(t, tc.wantDiff, diff)
-		})
-	}
-}
-
 func TestTable_GetFirstUniqueNotNullColumn(t *testing.T) {
 	t.Parallel()
 
diff --git a/pkg/wal/processor/search/store/search_store_test.go b/pkg/wal/processor/search/store/search_store_test.go
index ad755cf..9d6a8a5 100644
--- a/pkg/wal/processor/search/store/search_store_test.go
+++ b/pkg/wal/processor/search/store/search_store_test.go
@@ -856,7 +856,7 @@ func TestStore_updateMapping(t *testing.T) {
 	tests := []struct {
 		name   string
 		client searchstore.Client
-		diff   *schemalog.SchemaDiff
+		diff   *schemalog.Diff
 		mapper search.Mapper
 
 		wantErr error
@@ -902,9 +902,13 @@ func TestStore_updateMapping(t *testing.T) {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},
-			diff: &schemalog.SchemaDiff{
-				ColumnsToAdd: []schemalog.Column{
-					{Name: "col-1", PgstreamID: "pgstreamid-1"},
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						ColumnsAdded: []schemalog.Column{
+							{Name: "col-1", PgstreamID: "pgstreamid-1"},
+						},
+					},
 				},
 			},
 			mapper: &searchmocks.Mapper{
@@ -940,8 +944,8 @@ func TestStore_updateMapping(t *testing.T) {
 					return nil
 				},
 			},
-			diff: &schemalog.SchemaDiff{
-				TablesToRemove: []schemalog.Table{
+			diff: &schemalog.Diff{
+				TablesRemoved: []schemalog.Table{
 					{PgstreamID: "id-1"},
 					{PgstreamID: "id-2"},
 				},
@@ -965,9 +969,13 @@ func TestStore_updateMapping(t *testing.T) {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},
-			diff: &schemalog.SchemaDiff{
-				ColumnsToAdd: []schemalog.Column{
-					{Name: "col-1", PgstreamID: "pgstreamid-1"},
+			diff: &schemalog.Diff{
+				TablesChanged: []schemalog.TableDiff{
+					{
+						ColumnsAdded: []schemalog.Column{
+							{Name: "col-1", PgstreamID: "pgstreamid-1"},
+						},
+					},
 				},
 			},
 
@@ -989,8 +997,8 @@ func TestStore_updateMapping(t *testing.T) {
 					return errTest
 				},
 			},
-			diff: &schemalog.SchemaDiff{
-				TablesToRemove: []schemalog.Table{
+			diff: &schemalog.Diff{
+				TablesRemoved: []schemalog.Table{
 					{PgstreamID: "id-1"},
 					{PgstreamID: "id-2"},
 				},