diff --git a/connectors/cassandra/schema.go b/connectors/cassandra/schema.go
index 4401ad6a..4a5196cc 100644
--- a/connectors/cassandra/schema.go
+++ b/connectors/cassandra/schema.go
@@ -113,14 +113,16 @@ func checkClusteringKeys(ed *dosa.EntityDefinition, md *gocql.TableMetadata) err
 	}
 	return nil
 }
+
 func checkColumns(ed *dosa.EntityDefinition, md *gocql.TableMetadata, schemaErrors *RepairableSchemaMismatchError) {
 	// Check each column
 	for _, col := range ed.Columns {
-		_, ok := md.Columns[col.Name]
-		if !ok {
-			schemaErrors.MissingColumns = append(schemaErrors.MissingColumns, MissingColumn{Column: *col, Tablename: ed.Name})
+		cassandraCol, ok := md.Columns[col.Name]
+		if ok && cassandraCol.Type.Type().String() == cassandraType(col.Type) {
+			continue
 		}
-		// TODO: check column type
+
+		schemaErrors.MissingColumns = append(schemaErrors.MissingColumns, MissingColumn{Column: *col, Tablename: ed.Name})
 	}
 }
 
diff --git a/connectors/cassandra/schema_test.go b/connectors/cassandra/schema_test.go
index 4b9e2352..73128fd9 100644
--- a/connectors/cassandra/schema_test.go
+++ b/connectors/cassandra/schema_test.go
@@ -50,6 +50,35 @@ func TestCompareStructToSchemaWrongPk(t *testing.T) {
 	assert.Contains(t, err.Error(), `"test"`)
 }
 
+// test that the cluster key in gocql doesn't match cluster key in dosa definition. c1 != c2
+func TestCompareStructToSchemaWrongCk(t *testing.T) {
+	ed := dosa.EntityDefinition{
+		Key: &dosa.PrimaryKey{
+			PartitionKeys:  []string{"p1"},
+			ClusteringKeys: []*dosa.ClusteringKey{{Name: "c1", Descending: true}},
+		},
+		Name: "test",
+		Columns: []*dosa.ColumnDefinition{
+			{Name: "p1", Type: dosa.String},
+			{Name: "c1", Type: dosa.String},
+		},
+	}
+	md := gocql.TableMetadata{
+		PartitionKey: []*gocql.ColumnMetadata{
+			{Name: "p1", Type: TestType{typ: gocql.TypeText}},
+		},
+		ClusteringColumns: []*gocql.ColumnMetadata{
+			{Name: "c2", Type: TestType{typ: gocql.TypeText}},
+		},
+		Columns: map[string]*gocql.ColumnMetadata{
+			"p1": {Name: "p1", Type: TestType{typ: gocql.TypeText}},
+		}}
+	missing := RepairableSchemaMismatchError{}
+	err := compareStructToSchema(&ed, &md, &missing)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), `"test"`)
+}
+
 func TestCompareStructToSchemaMissingColumn(t *testing.T) {
 	ed := dosa.EntityDefinition{Key: &dosa.PrimaryKey{
 		PartitionKeys: []string{"p1"}},
@@ -60,14 +89,14 @@ func TestCompareStructToSchemaMissingColumn(t *testing.T) {
 		},
 	}
 	md := gocql.TableMetadata{PartitionKey: []*gocql.ColumnMetadata{
-		{Name: "p1", Type: TestType{typ: gocql.TypeVarchar}},
+		{Name: "p1", Type: TestType{typ: gocql.TypeText}},
 	},
 		Columns: map[string]*gocql.ColumnMetadata{
-			"p1": {Name: "p1", Type: TestType{typ: gocql.TypeVarchar}},
+			"p1": {Name: "p1", Type: TestType{typ: gocql.TypeText}},
 		}}
 	missing := RepairableSchemaMismatchError{}
 	err := compareStructToSchema(&ed, &md, &missing)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 	assert.True(t, missing.HasMissing())
 	assert.Equal(t, 1, len(missing.MissingColumns))
 	assert.Equal(t, "test", missing.MissingColumns[0].Tablename)
@@ -75,6 +104,55 @@ func TestCompareStructToSchemaMissingColumn(t *testing.T) {
 	assert.Contains(t, missing.Error(), "Missing 1 column")
 }
 
+// dosa string type is a gocql text type, not varchar
+func TestCompareStructToSchemaMissingColumnType(t *testing.T) {
+	ed := dosa.EntityDefinition{Key: &dosa.PrimaryKey{
+		PartitionKeys: []string{"pk"}},
+		Name: "test",
+		Columns: []*dosa.ColumnDefinition{
+			{Name: "pk", Type: dosa.String},
+		},
+	}
+	md := gocql.TableMetadata{
+		PartitionKey: []*gocql.ColumnMetadata{
+			{Name: "pk", Type: TestType{typ: gocql.TypeVarchar}},
+		},
+		Columns: map[string]*gocql.ColumnMetadata{
+			"pk": {Name: "pk", Type: TestType{typ: gocql.TypeVarchar}},
+		},
+	}
+	missing := RepairableSchemaMismatchError{}
+	err := compareStructToSchema(&ed, &md, &missing)
+	assert.NoError(t, err)
+	assert.True(t, missing.HasMissing())
+	assert.Equal(t, 1, len(missing.MissingColumns))
+	assert.Equal(t, "test", missing.MissingColumns[0].Tablename)
+	assert.Equal(t, "pk", missing.MissingColumns[0].Column.Name)
+	assert.Contains(t, missing.Error(), "Missing 1 column")
+}
+
+func TestCompareStructToSchema(t *testing.T) {
+	ed := dosa.EntityDefinition{Key: &dosa.PrimaryKey{
+		PartitionKeys: []string{"p1"}},
+		Name: "test",
+		Columns: []*dosa.ColumnDefinition{
+			{Name: "p1", Type: dosa.Int32},
+		},
+	}
+	md := gocql.TableMetadata{
+		PartitionKey: []*gocql.ColumnMetadata{
+			{Name: "p1", Type: TestType{typ: gocql.TypeInt}},
+		},
+		Columns: map[string]*gocql.ColumnMetadata{
+			"p1": {Name: "p1", Type: TestType{typ: gocql.TypeInt}},
+		},
+	}
+	missing := RepairableSchemaMismatchError{}
+	err := compareStructToSchema(&ed, &md, &missing)
+	assert.NoError(t, err)
+	assert.False(t, missing.HasMissing())
+}
+
 type TestType struct {
 	typ gocql.Type
 }