diff --git a/lib/src/main/java/io/cloudquery/schema/Table.java b/lib/src/main/java/io/cloudquery/schema/Table.java
index f900253..8f91c09 100644
--- a/lib/src/main/java/io/cloudquery/schema/Table.java
+++ b/lib/src/main/java/io/cloudquery/schema/Table.java
@@ -1,5 +1,9 @@
 package io.cloudquery.schema;
 
+import static io.cloudquery.schema.TableColumnChangeType.ADD;
+import static io.cloudquery.schema.TableColumnChangeType.REMOVE;
+import static io.cloudquery.schema.TableColumnChangeType.UPDATE;
+
 import io.cloudquery.glob.Glob;
 import io.cloudquery.schema.Column.ColumnBuilder;
 import io.cloudquery.transformers.TransformerException;
@@ -18,6 +22,7 @@
 @Builder(toBuilder = true)
 @Getter
 public class Table {
+
   public interface Transform {
     void transformTable(Table table) throws TransformerException;
   }
@@ -203,4 +208,39 @@ public Optional<Column> getColumn(String name) {
     }
     return Optional.empty();
   }
+
+  public List<TableColumnChange> getChanges(Table old) {
+    List<TableColumnChange> changes = new ArrayList<>();
+    for (Column currentColumn : columns) {
+      Optional<Column> oldColumn = old.getColumn(currentColumn.getName());
+      if (oldColumn.isEmpty()) {
+        changes.add(new TableColumnChange(ADD, currentColumn.getName(), currentColumn, null));
+        continue;
+      }
+      if (shouldUpdate(currentColumn, oldColumn.get())) {
+        changes.add(
+            new TableColumnChange(UPDATE, currentColumn.getName(), currentColumn, oldColumn.get()));
+      }
+    }
+    for (Column column : old.columns) {
+      Optional<Column> otherColumn = getColumn(column.getName());
+      if (otherColumn.isEmpty()) {
+        changes.add(new TableColumnChange(REMOVE, column.getName(), null, column));
+      }
+    }
+    return changes;
+  }
+
+  private static boolean shouldUpdate(Column currentColumn, Column oldColumn) {
+    if (!oldColumn.getType().equals(currentColumn.getType())) {
+      return true;
+    }
+    if (oldColumn.isPrimaryKey() != currentColumn.isPrimaryKey()) {
+      return true;
+    }
+    if (oldColumn.isNotNull() != currentColumn.isNotNull()) {
+      return true;
+    }
+    return oldColumn.isUnique() != currentColumn.isUnique();
+  }
 }
diff --git a/lib/src/main/java/io/cloudquery/schema/TableColumnChange.java b/lib/src/main/java/io/cloudquery/schema/TableColumnChange.java
new file mode 100644
index 0000000..8af4f13
--- /dev/null
+++ b/lib/src/main/java/io/cloudquery/schema/TableColumnChange.java
@@ -0,0 +1,13 @@
+package io.cloudquery.schema;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public class TableColumnChange {
+  private TableColumnChangeType type;
+  private String columnName;
+  private Column current;
+  private Column previous;
+}
diff --git a/lib/src/main/java/io/cloudquery/schema/TableColumnChangeType.java b/lib/src/main/java/io/cloudquery/schema/TableColumnChangeType.java
new file mode 100644
index 0000000..9948294
--- /dev/null
+++ b/lib/src/main/java/io/cloudquery/schema/TableColumnChangeType.java
@@ -0,0 +1,7 @@
+package io.cloudquery.schema;
+
+public enum TableColumnChangeType {
+  REMOVE,
+  UPDATE,
+  ADD
+}
diff --git a/lib/src/test/java/io/cloudquery/schema/TableColumnChangeTest.java b/lib/src/test/java/io/cloudquery/schema/TableColumnChangeTest.java
new file mode 100644
index 0000000..a7a339e
--- /dev/null
+++ b/lib/src/test/java/io/cloudquery/schema/TableColumnChangeTest.java
@@ -0,0 +1,104 @@
+package io.cloudquery.schema;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import io.cloudquery.schema.Table.TableBuilder;
+import java.util.List;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TableColumnChangeTest {
+
+  private TableBuilder tableBuilder;
+
+  @BeforeEach
+  public void setUp() {
+    tableBuilder =
+        Table.builder()
+            .name("test")
+            .columns(
+                List.of(
+                    Column.builder().name("basic_column").type(ArrowType.Bool.INSTANCE).build()));
+  }
+
+  @Test
+  public void shouldBeNoChangesWithIdenticalTables() {
+    Table old = tableBuilder.build();
+    Table current = tableBuilder.build();
+
+    List<TableColumnChange> changes = current.getChanges(old);
+
+    assertEquals(0, changes.size(), "should be no changes");
+  }
+
+  @Test
+  public void shouldAddColumnIfNewTableHasAdditionalColumn() {
+    Table old = tableBuilder.build();
+    Table current =
+        tableBuilder
+            .columns(
+                List.of(
+                    Column.builder().name("basic_column").type(ArrowType.Bool.INSTANCE).build(),
+                    Column.builder()
+                        .name("additional_column")
+                        .type(ArrowType.Bool.INSTANCE)
+                        .build()))
+            .build();
+
+    List<TableColumnChange> changes = current.getChanges(old);
+
+    assertEquals(1, changes.size(), "should be 1 change");
+    TableColumnChange change = changes.get(0);
+    assertEquals(TableColumnChangeType.ADD, change.getType());
+    assertEquals("additional_column", change.getColumnName());
+    assertEquals("additional_column", change.getCurrent().getName());
+  }
+
+  @Test
+  public void shouldRemoveColumnIfNewTableHasLessColumns() {
+    Table old =
+        Table.builder()
+            .name("test")
+            .columns(
+                List.of(
+                    Column.builder().name("basic_column").type(ArrowType.Bool.INSTANCE).build(),
+                    Column.builder()
+                        .name("additional_column")
+                        .type(ArrowType.Bool.INSTANCE)
+                        .build()))
+            .build();
+
+    Table current = tableBuilder.build();
+
+    List<TableColumnChange> changes = current.getChanges(old);
+
+    assertEquals(1, changes.size(), "should be 1 change");
+    TableColumnChange change = changes.get(0);
+    assertEquals(TableColumnChangeType.REMOVE, change.getType());
+    assertEquals("additional_column", change.getColumnName());
+    assertEquals("additional_column", change.getPrevious().getName());
+  }
+
+  @Test
+  public void shouldUpdateColumnIfNewTableHasUpdateChange() {
+    Table old = tableBuilder.build();
+    Table current =
+        tableBuilder
+            .columns(
+                List.of(
+                    Column.builder().name("basic_column").type(ArrowType.Utf8.INSTANCE).build()))
+            .build();
+
+    List<TableColumnChange> changes = current.getChanges(old);
+
+    assertEquals(1, changes.size(), "should be 1 change");
+    TableColumnChange change = changes.get(0);
+    assertEquals(TableColumnChangeType.UPDATE, change.getType());
+    assertEquals("basic_column", change.getColumnName());
+    assertEquals("basic_column", change.getCurrent().getName());
+    assertEquals("basic_column", change.getPrevious().getName());
+    assertEquals(ArrowType.Bool.INSTANCE, change.getPrevious().getType());
+    assertEquals(ArrowType.Utf8.INSTANCE, change.getCurrent().getType());
+  }
+}