Handle unconvertible CREATE TABLE column defintions (#548)

Column definitions in `CREATE TABLE` statements offer [several
options](https://www.postgresql.org/docs/current/sql-createtable.html),
most of which aren't currently representable by `pgroll` `Column`
definitions.

Add tests and code to ensure that `CREATE TABLE` statements containing
columns that use any of these unrepresentable options fall back to raw
SQL operations.

Part of #504

Author: andrew-farries

pkg/sql2pgroll/create_table.go

@@ -29,6 +29,9 @@ func convertCreateStmt(stmt *pgq.CreateStmt) (migrations.Operations, error) {
 			if err != nil {
 				return nil, fmt.Errorf("error converting column definition: %w", err)
 			}
+			if column == nil {
+				return nil, nil
+			}
 			columns = append(columns, *column)
 		default:
 			return nil, nil
@@ -80,6 +83,10 @@ func canConvertCreateStatement(stmt *pgq.CreateStmt) bool {
 }
 
 func convertColumnDef(col *pgq.ColumnDef) (*migrations.Column, error) {
+	if !canConvertColumnDef(col) {
+		return nil, nil
+	}
+
 	// Convert the column type
 	typeString, err := pgq.DeparseTypeName(col.TypeName)
 	if err != nil {
@@ -111,3 +118,21 @@ func convertColumnDef(col *pgq.ColumnDef) (*migrations.Column, error) {
 		Pk:       pk,
 	}, nil
 }
+
+// canConvertColumnDef returns true iff `col` can be converted to a pgroll
+// `Column` definition.
+func canConvertColumnDef(col *pgq.ColumnDef) bool {
+	switch {
+	// Column storage options are not supported
+	case col.GetStorageName() != "":
+		return false
+	// Column compression options are not supported
+	case col.GetCompression() != "":
+		return false
+	// Column collation options are not supported
+	case col.GetCollClause() != nil:
+		return false
+	default:
+		return true
+	}
+}

pkg/sql2pgroll/create_table_test.go

@@ -112,6 +112,15 @@ func TestUnconvertableCreateTableStatements(t *testing.T) {
 		// The LIKE clause is not supported
 		"CREATE TABLE foo(a int, LIKE bar)",
 		"CREATE TABLE foo(LIKE bar)",
+
+		// Column `STORAGE` options are not supported
+		"CREATE TABLE foo(a int STORAGE PLAIN)",
+
+		// Column compression options are not supported
+		"CREATE TABLE foo(a text COMPRESSION pglz)",
+
+		// Column collation is not supported
+		"CREATE TABLE foo(a text COLLATE en_US)",
 	}
 
 	for _, sql := range tests {