Add support for AlterTable::drop_columns

CHANGES.md

@@ -6,6 +6,7 @@
 - Runtime: Updated Fivetran SDK to commit `18e037c`
 - Implemented `AlterTableRecreateStatements` for recreating a table
   with a new schema, used for processing primary key column changes.
+- Added support for `AlterTable::drop_columns` operation.
 
 ## v0.0.3 - 2025-05-23
 - Dependencies: Updated to `sqlalchemy-cratedb==0.42.0.dev2`

src/cratedb_fivetran_destination/main.py

@@ -121,6 +121,7 @@ def AlterTable(self, request, context):
         columns_new: t.List[common_pb2.Column] = []
         columns_changed: t.List[common_pb2.Column] = []
         columns_common: t.List[common_pb2.Column] = []
+        columns_deleted: t.List[common_pb2.Column] = []
 
         for column in new_table.columns:
             column_old = columns_old.get(column.name)
@@ -138,6 +139,27 @@ def AlterTable(self, request, context):
                     columns_changed.append(column)
 
         table_info = self._table_info_from_request(request)
+
+        if request.drop_columns:
+            new_column_names = [column.name for column in new_table.columns]
+            for column in old_table.columns:
+                if column.name not in new_column_names:
+                    columns_deleted.append(column)
+            if columns_deleted:
+                amendments = []
+                for column in columns_deleted:
+                    amendments.append(f'DROP COLUMN "{column.name}"')
+                with self.engine.connect() as connection:
+                    connection.execute(
+                        sa.text(f"ALTER TABLE {table_info.fullname} {', '.join(amendments)}")
+                    )
+                log_message(
+                    LOG_INFO, f"AlterTable: Successfully altered table: {table_info.fullname}"
+                )
+            else:
+                log_message(LOG_INFO, "AlterTable (drop columns): Nothing changed")
+            return destination_sdk_pb2.AlterTableResponse(success=True)
+
         if pk_has_changed:
             log_message(
                 LOG_WARNING,

tests/data/fivetran_migrations_ddl/configuration.json

@@ -0,0 +1,11 @@
+{
+  "writerType": "Database",
+  "url": "crate://",
+  "database": "doc",
+  "table": "testdrive",
+  "host": "localhost",
+  "port": "4200",
+  "user": "crate",
+  "password": "",
+  "enableEncryption": "false"
+}

tests/data/fivetran_migrations_ddl/schema_migrations_input_ddl.json

@@ -0,0 +1,53 @@
+{
+  "create_table" : {
+    "transaction": {
+      "columns": {
+        "id": "INT",
+        "amount": "DOUBLE",
+        "desc": "STRING"
+      },
+      "primary_key": ["id"]
+    }
+  },
+  "ops" : [
+    {
+      "upsert": {
+        "transaction": [
+          {"id":1, "amount": 100.45, "desc": null},
+          {"id":2, "amount": 150.33, "desc": "two"},
+          {"id":3, "amount": 150.33, "desc": "two"},
+          {"id":4, "amount": 150.33, "desc": "two"},
+          {"id":10, "amount": 200, "desc": "three"},
+          {"id":20, "amount": 50, "desc": "money"}
+        ]
+      }
+    }
+  ],
+  "schema_migration" : [
+    {
+      "add_column": [
+        {
+          "table": "transaction",
+          "column": "operation_time",
+          "data_type": "UTC_DATETIME"
+        }
+      ],
+      "change_column_data_type": [
+        {
+          "table": "transaction",
+          "column": "amount",
+          "data_type": "STRING"
+        }
+      ],
+      "drop_column": [
+        {
+          "table": "transaction",
+          "column": "desc"
+        }
+      ]
+    }
+  ],
+  "describe_table" : [
+    "transaction"
+  ]
+}

tests/test_adapter.py

@@ -220,6 +220,45 @@ def test_api_alter_table_nothing_changed(engine, capsys):
     assert format_log_message("AlterTable: Nothing changed", newline=True) in out
 
 
+def test_api_alter_table_drop_column_nothing_changed(engine, capsys):
+    """
+    Invoke gRPC API method `AlterTable` with `drop_columns=True`, but nothing changed.
+    """
+    from cratedb_fivetran_destination.main import CrateDBDestinationImpl
+
+    destination = CrateDBDestinationImpl()
+
+    with engine.connect() as conn:
+        conn.execute(sa.text("CREATE TABLE testdrive.foo (id INT)"))
+
+    # Invoke gRPC API method under test.
+    table: common_pb2.Table = common_pb2.Table(
+        name="foo",
+        columns=[
+            common_pb2.Column(
+                name="id",
+                type=common_pb2.DataType.INT,
+                primary_key=False,
+            )
+        ],
+    )
+    config = {"url": "crate://"}
+    response = destination.AlterTable(
+        request=destination_sdk_pb2.AlterTableRequest(
+            table=table, schema_name="testdrive", configuration=config, drop_columns=True
+        ),
+        context=destination_sdk_pb2.AlterTableResponse(),
+    )
+
+    # Validate outcome.
+    assert response.success is True
+    assert response.warning.message == ""
+
+    # Check log output.
+    out, err = capsys.readouterr()
+    assert format_log_message("AlterTable (drop columns): Nothing changed", newline=True) in out
+
+
 def test_api_alter_table_change_primary_key_type(engine, capsys):
     """
     Invoke gRPC API method `AlterTable`, changing the type of the primary key.

tests/test_integration.py

@@ -140,6 +140,18 @@ def test_integration_fivetran(capfd, services):
     assert "Describe Table: composite_table" in err
 
 
+@pytest.mark.parametrize("services", ["./tests/data/fivetran_migrations_ddl"], indirect=True)
+def test_integration_fivetran_migrations_ddl(capfd, services):
+    """
+    Verify the Fivetran destination tester runs to completion with Fivetran test data.
+    """
+
+    # Read out stdout and stderr.
+    out, err = capfd.readouterr()
+
+    assert "Describe Table: transaction" in err
+
+
 @pytest.mark.parametrize("services", ["./tests/data/cratedb_canonical"], indirect=True)
 def test_integration_cratedb(capfd, services, engine):
     """