filter excluded columns out on delete

add schema check
1db96950527c77fe01bdaacfcea453cc4ea9acac

transfer_manager/go/pkg/transformer/registry/filter/filter_columns_transformer.go

@@ -166,7 +166,7 @@ func (f *FilterColumnsTransformer) getColumnIndexes(item abstract.ChangeItem, fi
 
 func (f *FilterColumnsTransformer) trimChangeItem(original abstract.ChangeItem, filteredColsIndexes []int, filteredSchema *filteredTableSchema) (abstract.ChangeItem, error) {
 	filteredColumnsCount := len(filteredColsIndexes)
-	if len(original.ColumnNames) == filteredColumnsCount {
+	if len(original.ColumnNames) == filteredColumnsCount && len(filteredSchema.Columns) == len(original.TableSchema.Columns()) {
 		return original, nil
 	}