Repository: cloudspannerecosystem/harbourbridge
Pull request #95 (2020-12-15 04:34:34)

feature: secondary index
Index key PR comment:  > This PR is for the new feature: Index keys addressing issue https://github.com/cloudspannerecosystem/harbourbridge/issues/21 , changes included in this PR:
> *`internal/convert.go` - print indexes along with create table.
> *`internal/mapping.go` - added GetSpannerIndexKeyName to convert source key name to spanner key name avoiding collision.  
> * `mysql/infoschema.go` - support for index key getIndexes function
> * `mysql/mysqldump.go` - support for index key, we process CREATE INDEX statements, along with indexes mentioned in CREATE TABLE and ALTER TABLE, we also convert UNIQUE constraint to UNIQUE index.
> * `mysql/toddl.go` - support for index key
> * `postgres/infoschema.go` - support for index key getIndexes function
> * `postgres/pg_dump.go` - support for index key, we process CREATE INDEX statements, we also convert UNIQUE constraint to UNIQUE index. 
> * `postgres/toddl.go` - support for index key

> Note: we convert all unique constraint to unique secondary index in spanner as mentioned [here](https://cloud.google.com/spanner/docs/migrating-postgres-spanner#indexes) . this applies to both postgres and mysql.
> Note: we skip ordering of keys in mysqldump (i.e, ascending/descending) as mysqldump parser is not able to parse this order.
> This PR is completed.



agasheesh: Just FYI, this pull request seem to be fixing the issue https://github.com/cloudspannerecosystem/harbourbridge/issues/21. Please feel free to link the same here to linked issues.
Pradip97: > Finished review (after conflict resolution). Other than the issue noted below, I'm ready to approve this PR.
> 
> As noted in the chat room, I ran into an issue: I think we need to ensure that index names and foreign key constraint names don't overlap. In other words, it looks like index names and foreign key constraints share the same namespace in Spanner.
> 
> I ran into this while doing a final round of testing on sakila. Here's a cut down version:
> 
> CREATE TABLE film_category (
> film_id INT64 NOT NULL,
> category_id INT64 NOT NULL,
> last_update TIMESTAMP NOT NULL
> ) PRIMARY KEY (film_id, category_id);
> 
> CREATE TABLE category (
> category_id INT64 NOT NULL,
> name STRING(25) NOT NULL,
> last_update TIMESTAMP NOT NULL
> ) PRIMARY KEY (category_id);
> 
> CREATE INDEX fk_film_category_category ON film_category (category_id);
> 
> ALTER TABLE film_category ADD CONSTRAINT fk_film_category_category FOREIGN KEY (category_id) REFERENCES category (category_id);
> 
> This last statement fails with "Duplicate name in schema: fk_film_category_category".



Patch:
diff --git a/internal/convert.go b/internal/convert.go
index d29118e..1e4e876 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -181,9 +181,11 @@ func (conv *Conv) GetDDL(c ddl.Config) []string {
 	if c.Tables {
 		for _, t := range tables {
 			ddl = append(ddl, conv.SpSchema[t].PrintCreateTable(c))
+			for _, index := range conv.SpSchema[t].Indexes {
+				ddl = append(ddl, index.PrintCreateIndex(c))
+			}
 		}
 	}
-
 	// Append foreign key constraints to DDL.
 	// We always use alter table statements for foreign key constraints.
 	// The alternative of putting foreign key constraints in-line as part of create
diff --git a/internal/mapping.go b/internal/mapping.go
index 187b64f..0bfb82f 100644
--- a/internal/mapping.go
+++ b/internal/mapping.go
@@ -131,7 +131,7 @@ func GetSpannerCols(conv *Conv, srcTable string, srcCols []string) ([]string, er
 	return spCols, nil
 }
 
-// GetSpannerKeyName maps source foreign key name to
+// ToSpannerForeignKey maps source foreign key name to
 // legal Spanner foreign key name.
 // If the srcKeyName is empty string we can just return
 // empty string without error.
@@ -144,27 +144,44 @@ func GetSpannerCols(conv *Conv, srcTable string, srcCols []string) ([]string, er
 // (across the database). But in some source databases, such as PostgreSQL,
 // they only have to be unique for a table. Hence we must map each source
 // constraint name to a unique spanner constraint name.
-func GetSpannerKeyName(srcKeyName string, schemaForeignKeys map[string]bool) string {
-	if srcKeyName == "" {
+func ToSpannerForeignKey(srcId string, used map[string]bool) string {
+	if srcId == "" {
 		return ""
 	}
-	spKeyName, _ := FixName(srcKeyName)
-	if _, found := schemaForeignKeys[spKeyName]; found {
+	return getSpannerId(srcId, used)
+}
+
+// ToSpannerIndexName maps source index name to legal Spanner index name.
+// We need to make sure of the following things:
+// a) the new index name is legal
+// b) the new index name doesn't clash with other Spanner
+//    index names
+// Note that index key constraint names in Spanner have to be globally unique
+// (across the database). But in some source databases, such as MySQL,
+// they only have to be unique for a table. Hence we must map each source
+// constraint name to a unique spanner constraint name.
+func ToSpannerIndexName(srcId string, used map[string]bool) string {
+	return getSpannerId(srcId, used)
+}
+
+func getSpannerId(srcId string, used map[string]bool) string {
+	spKeyName, _ := FixName(srcId)
+	if _, found := used[spKeyName]; found {
 		// spKeyName has been used before.
-		// Add unique postfix: use number of foreign keys so far.
+		// Add unique postfix: use number of keys so far.
 		// However, there is a chance this has already been used,
 		// so need to iterate.
-		id := len(schemaForeignKeys)
+		id := len(used)
 		for {
 			c := spKeyName + "_" + strconv.Itoa(id)
-			if _, found := schemaForeignKeys[c]; !found {
+			if _, found := used[c]; !found {
 				spKeyName = c
 				break
 			}
 			id++
 		}
 	}
-	schemaForeignKeys[spKeyName] = true
+	used[spKeyName] = true
 	return spKeyName
 }
 
diff --git a/mysql/README.md b/mysql/README.md
index a422141..ff177e7 100644
--- a/mysql/README.md
+++ b/mysql/README.md
@@ -199,6 +199,16 @@ and `ON UPDATE` actions, we drop them.
 Spanner does not currently support default values. We drop these
 MySQL features during conversion.
 
+### Secondary Indexes
+
+The tool maps MySQL secondary indexes to Spanner secondary indexes, and preserves
+constraint names where possible. Note that Spanner requires index key constraint
+names to be globally unique (within a database), but in MySQL they only have to be
+unique for a table, so we add a uniqueness suffix to a name if needed. The tool also
+maps `UNIQUE` constraint into `UNIQUE` secondary index. Note that due to limitations of our
+mysqldump parser, we are not able to handle key column ordering (i.e. ASC/DESC) in
+mysqldump files. All key columns in mysqldump files will be treated as ASC.
+
 ### Other MySQL features
 
 MySQL has many other features we haven't discussed, including functions,
diff --git a/mysql/infoschema.go b/mysql/infoschema.go
index 5b65121..123b320 100644
--- a/mysql/infoschema.go
+++ b/mysql/infoschema.go
@@ -208,6 +208,10 @@ func processTable(conv *internal.Conv, db *sql.DB, table schemaAndName) error {
 	if err != nil {
 		return fmt.Errorf("couldn't get foreign key constraints for table %s.%s: %s", table.schema, table.name, err)
 	}
+	indexes, err := getIndexes(conv, db, table)
+	if err != nil {
+		return fmt.Errorf("couldn't get indexes for table %s.%s: %s", table.schema, table.name, err)
+	}
 	colDefs, colNames := processColumns(conv, cols, constraints)
 	name := table.name
 	var schemaPKeys []schema.Key
@@ -219,6 +223,7 @@ func processTable(conv *internal.Conv, db *sql.DB, table schemaAndName) error {
 		ColNames:    colNames,
 		ColDefs:     colDefs,
 		PrimaryKeys: schemaPKeys,
+		Indexes:     indexes,
 		ForeignKeys: foreignKeys}
 	return nil
 }
@@ -375,6 +380,42 @@ func getForeignKeys(conv *internal.Conv, db *sql.DB, table schemaAndName) (forei
 	return foreignKeys, nil
 }
 
+// getIndexes return a list of all indexes for the specified table.
+func getIndexes(conv *internal.Conv, db *sql.DB, table schemaAndName) ([]schema.Index, error) {
+	q := `SELECT DISTINCT INDEX_NAME,COLUMN_NAME,SEQ_IN_INDEX,COLLATION,NON_UNIQUE
+		FROM INFORMATION_SCHEMA.STATISTICS 
+		WHERE TABLE_SCHEMA = ?
+			AND TABLE_NAME = ?
+			AND INDEX_NAME != 'PRIMARY' 
+		ORDER BY INDEX_NAME, SEQ_IN_INDEX;`
+	rows, err := db.Query(q, table.schema, table.name)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var name, column, sequence, nonUnique string
+	var collation sql.NullString
+	indexMap := make(map[string]schema.Index)
+	var indexNames []string
+	var indexes []schema.Index
+	for rows.Next() {
+		if err := rows.Scan(&name, &column, &sequence, &collation, &nonUnique); err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		if _, found := indexMap[name]; !found {
+			indexNames = append(indexNames, name)
+			indexMap[name] = schema.Index{Name: name, Unique: (nonUnique == "0")}
+		}
+		index := indexMap[name]
+		index.Keys = append(index.Keys, schema.Key{Column: column, Desc: (collation.Valid && collation.String == "D")})
+		indexMap[name] = index
+	}
+	for _, k := range indexNames {
+		indexes = append(indexes, indexMap[k])
+	}
+	return indexes, nil
+}
 func toType(dataType string, columnType string, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
 	switch {
 	case dataType == "set":
diff --git a/mysql/mysqldump.go b/mysql/mysqldump.go
index 0b541ff..60cb17f 100644
--- a/mysql/mysqldump.go
+++ b/mysql/mysqldump.go
@@ -142,12 +142,40 @@ func processStatement(conv *internal.Conv, stmt ast.StmtNode) bool {
 	case *ast.InsertStmt:
 		processInsertStmt(conv, s)
 		return true
+	case *ast.CreateIndexStmt:
+		if conv.SchemaMode() {
+			processCreateIndex(conv, s)
+		}
 	default:
 		conv.SkipStatement(NodeType(stmt))
 	}
 	return false
 }
 
+func processCreateIndex(conv *internal.Conv, stmt *ast.CreateIndexStmt) {
+	if stmt.Table == nil {
+		logStmtError(conv, stmt, fmt.Errorf("cannot process index statement with nil table."))
+		return
+	}
+	tableName, err := getTableName(stmt.Table)
+	if err != nil {
+		logStmtError(conv, stmt, fmt.Errorf("can't get table name: %w", err))
+		return
+	}
+	if _, ok := conv.SrcSchema[tableName]; ok {
+		ctable := conv.SrcSchema[tableName]
+		ctable.Indexes = append(ctable.Indexes, schema.Index{
+			Name:   stmt.IndexName,
+			Unique: (stmt.KeyType == ast.IndexKeyTypeUnique),
+			Keys:   toSchemaKeys(stmt.IndexPartSpecifications),
+		})
+		conv.SrcSchema[tableName] = ctable
+	} else {
+		conv.Unexpected(fmt.Sprintf("Table %s not found while processing index statement", tableName))
+		conv.SkipStatement(NodeType(stmt))
+	}
+}
+
 func processSetStmt(conv *internal.Conv, stmt *ast.SetStmt) {
 	if stmt.Variables != nil && len(stmt.Variables) > 0 {
 		for _, variable := range stmt.Variables {
@@ -187,6 +215,7 @@ func processCreateTable(conv *internal.Conv, stmt *ast.CreateTableStmt) {
 	colDef := make(map[string]schema.Column)
 	var keys []schema.Key
 	var fkeys []schema.ForeignKey
+	var index []schema.Index
 	for _, element := range stmt.Cols {
 		colname, col, constraint, err := processColumn(conv, tableName, element)
 		if err != nil {
@@ -201,6 +230,14 @@ func processCreateTable(conv *internal.Conv, stmt *ast.CreateTableStmt) {
 		if constraint.fk.Columns != nil {
 			fkeys = append(fkeys, constraint.fk)
 		}
+		if constraint.isUniqueKey {
+			// Convert unique column constraint in MySQL to a corresponding unique index in Spanner since
+			// Spanner doesn't support unique constraints on columns.
+			// TODO: Avoid Spanner-specific schema transformations in this file -- they should only
+			// appear in toddl.go. This file should focus on generic transformation from source
+			// database schemas into schema.go.
+			index = append(index, schema.Index{Name: "", Unique: true, Keys: []schema.Key{schema.Key{Column: colname, Desc: false}}})
+		}
 	}
 	conv.SchemaStatement(NodeType(stmt))
 	conv.SrcSchema[tableName] = schema.Table{
@@ -208,7 +245,8 @@ func processCreateTable(conv *internal.Conv, stmt *ast.CreateTableStmt) {
 		ColNames:    colNames,
 		ColDefs:     colDef,
 		PrimaryKeys: keys,
-		ForeignKeys: fkeys}
+		ForeignKeys: fkeys,
+		Indexes:     index}
 	for _, constraint := range stmt.Constraints {
 		processConstraint(conv, tableName, constraint, "CREATE TABLE")
 	}
@@ -228,18 +266,31 @@ func processConstraint(conv *internal.Conv, table string, constraint *ast.Constr
 		updateCols(conv, ast.ConstraintPrimaryKey, constraint.Keys, st.ColDefs, table)
 	case ast.ConstraintForeignKey:
 		st.ForeignKeys = append(st.ForeignKeys, toForeignKeys(conv, constraint))
+	case ast.ConstraintIndex:
+		st.Indexes = append(st.Indexes, schema.Index{Name: constraint.Name, Keys: toSchemaKeys(constraint.Keys)})
+	case ast.ConstraintUniq:
+		// Convert unique column constraint in MySQL to a corresponding unique index in Spanner since
+		// Spanner doesn't support unique constraints on columns.
+		// TODO: Avoid Spanner-specific schema transformations in this file -- they should only
+		// appear in toddl.go. This file should focus on generic transformation from source
+		// database schemas into schema.go.
+		st.Indexes = append(st.Indexes, schema.Index{Name: constraint.Name, Unique: true, Keys: toSchemaKeys(constraint.Keys)})
 	default:
 		updateCols(conv, ct, constraint.Keys, st.ColDefs, table)
 	}
 	conv.SrcSchema[table] = st
 }
 
-// toSchemaKeys converts a string list of MySQL primary keys to
-// schema primary keys.
+// toSchemaKeys converts a string list of MySQL keys to schema keys.
+// Note that we map all MySQL keys to ascending ordered schema keys.
+// For primary keys: this is fine because MySQL primary keys are always ascending.
+// However, for non-primary keys (aka indexes) this is incorrect: we are dropping
+// the MySQL key order specification, as mysqldump parser is not able to parse the
+// order. Check this for more details:
+// https://github.com/cloudspannerecosystem/harbourbridge/issues/96
+// TODO: Resolve ordering issue for non-primary keys.
 func toSchemaKeys(columns []*ast.IndexPartSpecification) (keys []schema.Key) {
 	for _, colname := range columns {
-		// MySQL primary keys have no notation of ascending/descending.
-		// We map them all into ascending primary keys.
 		keys = append(keys, schema.Key{Column: colname.Column.Name.String()})
 	}
 	return keys
@@ -321,6 +372,16 @@ func processAlterTable(conv *internal.Conv, stmt *ast.AlterTableStmt) {
 					ctable.ForeignKeys = append(ctable.ForeignKeys, constraint.fk)
 					conv.SrcSchema[tableName] = ctable
 				}
+				if constraint.isUniqueKey {
+					// Convert unique column constraint in mysql to a corresponding unique index in Spanner since
+					// Spanner doesn't support unique constraints on columns.
+					// TODO: Avoid Spanner-specific schema transformations in this file -- they should only
+					// appear in toddl.go. This file should focus on generic transformation from source
+					// database schemas into schema.go.
+					ctable := conv.SrcSchema[tableName]
+					ctable.Indexes = append(ctable.Indexes, schema.Index{Name: "", Unique: true, Keys: []schema.Key{schema.Key{Column: colname, Desc: false}}})
+					conv.SrcSchema[tableName] = ctable
+				}
 				conv.SchemaStatement(NodeType(stmt))
 			default:
 				conv.SkipStatement(NodeType(stmt))
@@ -373,8 +434,9 @@ func processColumn(conv *internal.Conv, tableName string, col *ast.ColumnDef) (s
 }
 
 type columnConstraint struct {
-	isPk bool
-	fk   schema.ForeignKey
+	isPk        bool
+	isUniqueKey bool
+	fk          schema.ForeignKey
 }
 
 // updateColsByOption is specifially for ColDef constraints.
@@ -405,6 +467,7 @@ func updateColsByOption(conv *internal.Conv, tableName string, col *ast.ColumnDe
 			}
 		case ast.ColumnOptionUniqKey:
 			column.Unique = true
+			cc.isUniqueKey = true
 		case ast.ColumnOptionCheck:
 			column.Ignored.Check = true
 		case ast.ColumnOptionReference:
diff --git a/mysql/toddl.go b/mysql/toddl.go
index d7a5fd4..14ce3f9 100644
--- a/mysql/toddl.go
+++ b/mysql/toddl.go
@@ -32,7 +32,23 @@ import (
 // Spanner. It uses the source schema in conv.SrcSchema, and writes
 // the Spanner schema to conv.SpSchema.
 func schemaToDDL(conv *internal.Conv) error {
-	schemaForeignKeys := make(map[string]bool)
+	// Tracks Spanner names that have been used for foreign key constraints
+	// and indexes. We use this to ensure we generate unique names when
+	// we map from MySQL to Spanner since Spanner requires all foreign
+	// key and index names to be distinct (you can't use the same name
+	// for a foreign key constraint and an index).
+	usedNames := make(map[string]bool)
+	// As Spanner uses same namespace for table names, foreign key constraint
+	// names and index names, we need to pre-populate usedNames with Spanner table
+	// names to handle collision with foreign key names and index names.
+	for _, srcTable := range conv.SrcSchema {
+		spTableName, err := internal.GetSpannerTable(conv, srcTable.Name)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Couldn't map source table %s to Spanner: %s", srcTable.Name, err))
+			continue
+		}
+		usedNames[spTableName] = true
+	}
 	for _, srcTable := range conv.SrcSchema {
 		spTableName, err := internal.GetSpannerTable(conv, srcTable.Name)
 		if err != nil {
@@ -83,7 +99,8 @@ func schemaToDDL(conv *internal.Conv) error {
 			ColNames: spColNames,
 			ColDefs:  spColDef,
 			Pks:      cvtPrimaryKeys(conv, srcTable.Name, srcTable.PrimaryKeys),
-			Fks:      cvtForeignKeys(conv, srcTable.Name, srcTable.ForeignKeys, schemaForeignKeys),
+			Fks:      cvtForeignKeys(conv, srcTable.Name, srcTable.ForeignKeys, usedNames),
+			Indexes:  cvtIndexes(conv, spTableName, srcTable.Name, srcTable.Indexes, usedNames),
 			Comment:  comment}
 	}
 	internal.ResolveRefs(conv)
@@ -196,7 +213,7 @@ func cvtPrimaryKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Key)
 	return spKeys
 }
 
-func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.ForeignKey, schemaForeignKeys map[string]bool) []ddl.Foreignkey {
+func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.ForeignKey, usedNames map[string]bool) []ddl.Foreignkey {
 	var spKeys []ddl.Foreignkey
 	for _, key := range srcKeys {
 		if len(key.Columns) != len(key.ReferColumns) {
@@ -219,7 +236,7 @@ func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Forei
 			spCols = append(spCols, spCol)
 			spReferCols = append(spReferCols, spReferCol)
 		}
-		spKeyName := internal.GetSpannerKeyName(key.Name, schemaForeignKeys)
+		spKeyName := internal.ToSpannerForeignKey(key.Name, usedNames)
 		spKey := ddl.Foreignkey{
 			Name:         spKeyName,
 			Columns:      spCols,
@@ -229,3 +246,32 @@ func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Forei
 	}
 	return spKeys
 }
+
+func cvtIndexes(conv *internal.Conv, spTableName string, srcTable string, srcIndexes []schema.Index, usedNames map[string]bool) []ddl.CreateIndex {
+	var spIndexes []ddl.CreateIndex
+	for _, srcIndex := range srcIndexes {
+		var spKeys []ddl.IndexKey
+		for _, k := range srcIndex.Keys {
+			spCol, err := internal.GetSpannerCol(conv, srcTable, k.Column, true)
+			if err != nil {
+				conv.Unexpected(fmt.Sprintf("Can't map index key column name for table %s", srcTable))
+				continue
+			}
+			spKeys = append(spKeys, ddl.IndexKey{Col: spCol, Desc: k.Desc})
+		}
+		if srcIndex.Name == "" {
+			// Generate a name if index name is empty in MySQL.
+			// Collision of index name will be handled by ToSpannerIndexName.
+			srcIndex.Name = fmt.Sprintf("Index_%s", srcTable)
+		}
+		spIndexName := internal.ToSpannerIndexName(srcIndex.Name, usedNames)
+		spIndex := ddl.CreateIndex{
+			Name:   spIndexName,
+			Table:  spTableName,
+			Unique: srcIndex.Unique,
+			Keys:   spKeys,
+		}
+		spIndexes = append(spIndexes, spIndex)
+	}
+	return spIndexes
+}
diff --git a/postgres/README.md b/postgres/README.md
index 36aeed5..5e2dae7 100644
--- a/postgres/README.md
+++ b/postgres/README.md
@@ -201,6 +201,13 @@ Spanner doesn't support `ON DELETE` and `ON UPDATE` actions, so we drop these.
 Spanner does not currently support default values. We drop these
 PostgreSQL features during conversion.
 
+### Secondary Indexes
+
+The tool maps PostgresSQL secondary indexes to Spanner secondary indexes, preserving
+constraint names where possible. The tool also maps PostgreSQL `UNIQUE` constraints to
+Spanner `UNIQUE` secondary indexes. Check [here](https://cloud.google.com/spanner/docs/migrating-postgres-spanner#indexes)
+for more details.
+
 ### Other PostgreSQL features
 
 PostgreSQL has many other features we haven't discussed, including functions,
diff --git a/postgres/infoschema.go b/postgres/infoschema.go
index a15ceb5..f8c47e2 100644
--- a/postgres/infoschema.go
+++ b/postgres/infoschema.go
@@ -240,6 +240,10 @@ func processTable(conv *internal.Conv, db *sql.DB, table schemaAndName) error {
 	if err != nil {
 		return fmt.Errorf("couldn't get foreign key constraints for table %s.%s: %s", table.schema, table.name, err)
 	}
+	indexes, err := getIndexes(conv, db, table)
+	if err != nil {
+		return fmt.Errorf("couldn't get indexes for table %s.%s: %s", table.schema, table.name, err)
+	}
 	colDefs, colNames := processColumns(conv, cols, constraints)
 	name := buildTableName(table.schema, table.name)
 	var schemaPKeys []schema.Key
@@ -251,6 +255,7 @@ func processTable(conv *internal.Conv, db *sql.DB, table schemaAndName) error {
 		ColNames:    colNames,
 		ColDefs:     colDefs,
 		PrimaryKeys: schemaPKeys,
+		Indexes:     indexes,
 		ForeignKeys: foreignKeys}
 	return nil
 }
@@ -415,6 +420,68 @@ func getForeignKeys(conv *internal.Conv, db *sql.DB, table schemaAndName) (forei
 	return foreignKeys, nil
 }
 
+// getIndexes return a list of all indexes for the specified table.
+// Note: Extracting index definitions from PostgreSQL information schema tables is complex.
+// See https://stackoverflow.com/questions/6777456/list-all-index-names-column-names-and-its-table-name-of-a-postgresql-database/44460269#44460269
+// for background.
+func getIndexes(conv *internal.Conv, db *sql.DB, table schemaAndName) ([]schema.Index, error) {
+	q := `SELECT
+			irel.relname AS index_name,
+			a.attname AS column_name,
+			1 + Array_position(i.indkey, a.attnum) AS column_position,
+			i.indisunique AS is_unique,
+			CASE o.OPTION & 1 WHEN 1 THEN 'DESC' ELSE 'ASC' END AS order
+		FROM pg_index AS i
+		JOIN pg_class AS trel
+		ON trel.oid = i.indrelid
+		JOIN pg_namespace AS tnsp
+		ON trel.relnamespace = tnsp.oid
+		JOIN pg_class AS irel
+		ON irel.oid = i.indexrelid
+		CROSS JOIN LATERAL UNNEST (i.indkey) WITH ordinality AS c (colnum, ordinality)
+		LEFT JOIN LATERAL UNNEST (i.indoption) WITH ordinality AS o (OPTION, ordinality)
+		ON c.ordinality = o.ordinality
+		JOIN pg_attribute AS a
+		ON trel.oid = a.attrelid
+			AND a.attnum = c.colnum
+		WHERE tnsp.nspname= $1
+			AND trel.relname= $2
+			AND i.indisprimary = false
+		GROUP BY tnsp.nspname,
+           		trel.relname,
+           		irel.relname,
+           		a.attname,
+           		array_position(i.indkey, a.attnum),
+           		o.OPTION,i.indisunique
+		ORDER BY irel.relname, array_position(i.indkey, a.attnum);`
+	rows, err := db.Query(q, table.schema, table.name)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var name, column, sequence, isUnique, collation string
+	indexMap := make(map[string]schema.Index)
+	var indexNames []string
+	var indexes []schema.Index
+	for rows.Next() {
+		if err := rows.Scan(&name, &column, &sequence, &isUnique, &collation); err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		if _, found := indexMap[name]; !found {
+			indexNames = append(indexNames, name)
+			indexMap[name] = schema.Index{Name: name, Unique: (isUnique == "true")}
+		}
+		index := indexMap[name]
+		index.Keys = append(index.Keys, schema.Key{Column: column, Desc: (collation == "DESC")})
+		indexMap[name] = index
+	}
+	for _, k := range indexNames {
+		indexes = append(indexes, indexMap[k])
+	}
+	return indexes, nil
+}
+
 func toType(dataType string, elementDataType sql.NullString, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
 	switch {
 	case dataType == "ARRAY" && elementDataType.Valid:
diff --git a/postgres/pgdump.go b/postgres/pgdump.go
index 2619a29..f434d63 100644
--- a/postgres/pgdump.go
+++ b/postgres/pgdump.go
@@ -185,6 +185,10 @@ func processStatements(conv *internal.Conv, statements []nodes.Node) *copyOrInse
 			if conv.SchemaMode() {
 				processVariableSetStmt(conv, n)
 			}
+		case nodes.IndexStmt:
+			if conv.SchemaMode() {
+				processIndexStmt(conv, n)
+			}
 		default:
 			conv.SkipStatement(prNodes([]nodes.Node{node}))
 		}
@@ -192,6 +196,29 @@ func processStatements(conv *internal.Conv, statements []nodes.Node) *copyOrInse
 	return nil
 }
 
+func processIndexStmt(conv *internal.Conv, n nodes.IndexStmt) {
+	if n.Relation == nil {
+		logStmtError(conv, n, fmt.Errorf("cannot process index statement with nil relation."))
+		return
+	}
+	tableName, err := getTableName(conv, *n.Relation)
+	if err != nil {
+		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
+		return
+	}
+	if ctable, ok := conv.SrcSchema[tableName]; ok {
+		ctable.Indexes = append(ctable.Indexes, schema.Index{
+			Name:   *n.Idxname,
+			Unique: n.Unique,
+			Keys:   toIndexKeys(n.IndexParams.Items),
+		})
+		conv.SrcSchema[tableName] = ctable
+	} else {
+		conv.Unexpected(fmt.Sprintf("Table %s not found while processing index statement", tableName))
+		conv.SkipStatement(prNodes([]nodes.Node{n}))
+	}
+}
+
 func processAlterTableStmt(conv *internal.Conv, n nodes.AlterTableStmt) {
 	if n.Relation == nil {
 		logStmtError(conv, n, fmt.Errorf("relation is nil"))
@@ -485,8 +512,8 @@ func getTableName(conv *internal.Conv, n nodes.RangeVar) (string, error) {
 type constraint struct {
 	ct   nodes.ConstrType
 	cols []string
+	name string // Used for FOREIGN KEY or SECONDARY INDEX
 	/* Fields used for FOREIGN KEY constraints: */
-	name       string
 	referCols  []string
 	referTable string
 }
@@ -499,7 +526,7 @@ func extractConstraints(conv *internal.Conv, n nodes.Node, table string, l []nod
 		case nodes.Constraint:
 			var cols, referCols []string
 			var referTable string
-			var fkName string
+			var conName string
 			switch d.Contype {
 			case nodes.CONSTR_FOREIGN:
 				t, err := getTableName(conv, *d.Pktable)
@@ -510,7 +537,7 @@ func extractConstraints(conv *internal.Conv, n nodes.Node, table string, l []nod
 				}
 				referTable = t
 				if d.Conname != nil {
-					fkName = *d.Conname
+					conName = *d.Conname
 				}
 				for i := range d.FkAttrs.Items {
 					k, err := getString(d.FkAttrs.Items[i])
@@ -531,6 +558,9 @@ func extractConstraints(conv *internal.Conv, n nodes.Node, table string, l []nod
 					referCols = append(referCols, f)
 				}
 			default:
+				if d.Conname != nil {
+					conName = *d.Conname
+				}
 				for _, j := range d.Keys.Items {
 					k, err := getString(j)
 					if err != nil {
@@ -541,7 +571,7 @@ func extractConstraints(conv *internal.Conv, n nodes.Node, table string, l []nod
 					cols = append(cols, k)
 				}
 			}
-			cs = append(cs, constraint{ct: d.Contype, cols: cols, name: fkName, referCols: referCols, referTable: referTable})
+			cs = append(cs, constraint{ct: d.Contype, cols: cols, name: conName, referCols: referCols, referTable: referTable})
 		default:
 			conv.Unexpected(fmt.Sprintf("Processing %v statement: found %s node while processing constraints\n", reflect.TypeOf(n), reflect.TypeOf(d)))
 		}
@@ -585,6 +615,15 @@ func updateSchema(conv *internal.Conv, table string, cs []constraint, stmtType s
 			ct := conv.SrcSchema[table]
 			ct.ForeignKeys = append(ct.ForeignKeys, toForeignKeys(c)) // Append to previous foreign keys.
 			conv.SrcSchema[table] = ct
+		case nodes.CONSTR_UNIQUE:
+			// Convert unique column constraint in postgres to a corresponding unique index in Spanner since
+			// Spanner doesn't support unique constraints on columns.
+			// TODO: Avoid Spanner-specific schema transformations in this file -- they should only
+			// appear in toddl.go. This file should focus on generic transformation from source
+			// database schemas into schema.go.
+			ct := conv.SrcSchema[table]
+			ct.Indexes = append(ct.Indexes, schema.Index{Name: c.name, Unique: true, Keys: toSchemaKeys(conv, table, c.cols)})
+			conv.SrcSchema[table] = ct
 		default:
 			ct := conv.SrcSchema[table]
 			updateCols(c.ct, c.cols, ct.ColDefs)
@@ -620,6 +659,19 @@ func toSchemaKeys(conv *internal.Conv, table string, s []string) (l []schema.Key
 	return l
 }
 
+// toIndexKeys converts a list of PostgreSQL index keys to schema index keys.
+func toIndexKeys(s []nodes.Node) []schema.Key {
+	var l []schema.Key
+	for _, k := range s {
+		desc := false
+		if k.(nodes.IndexElem).Ordering == nodes.SORTBY_DESC {
+			desc = true
+		}
+		l = append(l, schema.Key{Column: *k.(nodes.IndexElem).Name, Desc: desc})
+	}
+	return l
+}
+
 // toForeignKeys converts a string list of PostgreSQL foreign keys to
 // schema foreign keys.
 func toForeignKeys(fk constraint) (fkey schema.ForeignKey) {
diff --git a/postgres/toddl.go b/postgres/toddl.go
index 401f6cc..2463eff 100644
--- a/postgres/toddl.go
+++ b/postgres/toddl.go
@@ -28,7 +28,23 @@ import (
 // Spanner. It uses the source schema in conv.SrcSchema, and writes
 // the Spanner schema to conv.SpSchema.
 func schemaToDDL(conv *internal.Conv) error {
-	schemaForeignKeys := make(map[string]bool)
+	// Tracks Spanner names that have been used for foreign key constraints
+	// and indexes. We use this to ensure we generate unique names when
+	// we map from Postgres to Spanner since Spanner requires all foreign
+	// key and index names to be distinct (you can't use the same name
+	// for a foreign key constraint and an index).
+	usedNames := make(map[string]bool)
+	// As Spanner uses same namespace for table names, foreign key constraint
+	// names and index names, we need to pre-populate usedNames with Spanner table
+	// names to handle collision with foreign key names and index names.
+	for _, srcTable := range conv.SrcSchema {
+		spTableName, err := internal.GetSpannerTable(conv, srcTable.Name)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Couldn't map source table %s to Spanner: %s", srcTable.Name, err))
+			continue
+		}
+		usedNames[spTableName] = true
+	}
 	for _, srcTable := range conv.SrcSchema {
 		spTableName, err := internal.GetSpannerTable(conv, srcTable.Name)
 		if err != nil {
@@ -76,7 +92,8 @@ func schemaToDDL(conv *internal.Conv) error {
 			ColNames: spColNames,
 			ColDefs:  spColDef,
 			Pks:      cvtPrimaryKeys(conv, srcTable.Name, srcTable.PrimaryKeys),
-			Fks:      cvtForeignKeys(conv, srcTable.Name, srcTable.ForeignKeys, schemaForeignKeys),
+			Fks:      cvtForeignKeys(conv, srcTable.Name, srcTable.ForeignKeys, usedNames),
+			Indexes:  cvtIndexes(conv, spTableName, srcTable.Name, srcTable.Indexes, usedNames),
 			Comment:  comment}
 	}
 	internal.ResolveRefs(conv)
@@ -192,7 +209,7 @@ func cvtPrimaryKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Key)
 	return spKeys
 }
 
-func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.ForeignKey, schemaForeignKeys map[string]bool) []ddl.Foreignkey {
+func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.ForeignKey, usedNames map[string]bool) []ddl.Foreignkey {
 	var spKeys []ddl.Foreignkey
 	for _, key := range srcKeys {
 		if len(key.Columns) != len(key.ReferColumns) {
@@ -215,7 +232,7 @@ func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Forei
 			spCols = append(spCols, spCol)
 			spReferCols = append(spReferCols, spReferCol)
 		}
-		spKeyName := internal.GetSpannerKeyName(key.Name, schemaForeignKeys)
+		spKeyName := internal.ToSpannerForeignKey(key.Name, usedNames)
 		spKey := ddl.Foreignkey{
 			Name:         spKeyName,
 			Columns:      spCols,
@@ -225,3 +242,32 @@ func cvtForeignKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Forei
 	}
 	return spKeys
 }
+
+func cvtIndexes(conv *internal.Conv, spTableName string, srcTable string, srcIndexes []schema.Index, usedNames map[string]bool) []ddl.CreateIndex {
+	var spIndexes []ddl.CreateIndex
+	for _, srcIndex := range srcIndexes {
+		var spKeys []ddl.IndexKey
+		for _, k := range srcIndex.Keys {
+			spCol, err := internal.GetSpannerCol(conv, srcTable, k.Column, true)
+			if err != nil {
+				conv.Unexpected(fmt.Sprintf("Can't map index key column name for table %s", srcTable))
+				continue
+			}
+			spKeys = append(spKeys, ddl.IndexKey{Col: spCol, Desc: k.Desc})
+		}
+		if srcIndex.Name == "" {
+			// Generate a name if index name is empty in Postgres.
+			// Collision of index name will be handled by ToSpannerIndexName.
+			srcIndex.Name = fmt.Sprintf("Index_%s", srcTable)
+		}
+		spIndexName := internal.ToSpannerIndexName(srcIndex.Name, usedNames)
+		spIndex := ddl.CreateIndex{
+			Name:   spIndexName,
+			Table:  spTableName,
+			Unique: srcIndex.Unique,
+			Keys:   spKeys,
+		}
+		spIndexes = append(spIndexes, spIndex)
+	}
+	return spIndexes
+}
diff --git a/schema/schema.go b/schema/schema.go
index 5fe3f84..3c53127 100644
--- a/schema/schema.go
+++ b/schema/schema.go
@@ -74,8 +74,9 @@ type Key struct {
 
 // Index represents a database index.
 type Index struct {
-	Name string
-	Keys []Key
+	Name   string
+	Unique bool
+	Keys   []Key
 }
 
 // Type represents the type of a column.
diff --git a/spanner/ddl/ast.go b/spanner/ddl/ast.go
index c90acdd..d849ee8 100644
--- a/spanner/ddl/ast.go
+++ b/spanner/ddl/ast.go
@@ -168,6 +168,7 @@ type CreateTable struct {
 	ColDefs  map[string]ColumnDef // Provides definition of columns (a map for simpler/faster lookup during type processing)
 	Pks      []IndexKey
 	Fks      []Foreignkey
+	Indexes  []CreateIndex
 	Comment  string
 }
 
@@ -208,10 +209,11 @@ func (ct CreateTable) PrintCreateTable(config Config) string {
 // CreateIndex encodes the following DDL definition:
 //     create index: CREATE [UNIQUE] [NULL_FILTERED] INDEX index_name ON table_name ( key_part [, ...] ) [ storing_clause ] [ , interleave_clause ]
 type CreateIndex struct {
-	Name  string
-	Table string
-	Keys  []IndexKey
-	// We have no requirements for unique and null-filtered options and
+	Name   string
+	Table  string
+	Unique bool
+	Keys   []IndexKey
+	// We have no requirements for null-filtered option and
 	// storing/interleaving clauses yet, so we omit them for now.
 }
 
@@ -221,7 +223,11 @@ func (ci CreateIndex) PrintCreateIndex(c Config) string {
 	for _, p := range ci.Keys {
 		keys = append(keys, p.PrintIndexKey(c))
 	}
-	return fmt.Sprintf("CREATE INDEX %s ON %s (%s)", c.quote(ci.Name), c.quote(ci.Table), strings.Join(keys, ", "))
+	var unique string
+	if ci.Unique == true {
+		unique = "UNIQUE "
+	}
+	return fmt.Sprintf("CREATE %sINDEX %s ON %s (%s)", unique, c.quote(ci.Name), c.quote(ci.Table), strings.Join(keys, ", "))
 }
 
 // PrintForeignKeyAlterTable unparses the foreign keys using ALTER TABLE.


Test patch:
diff --git a/internal/convert_test.go b/internal/convert_test.go
index c93109c..e00a097 100644
--- a/internal/convert_test.go
+++ b/internal/convert_test.go
@@ -51,8 +51,9 @@ func TestGetDDL(t *testing.T) {
 			"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
 			"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Int64}},
 		},
-		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
-		Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk1", Columns: []string{"b"}, ReferTable: "ref_table1", ReferColumns: []string{"ref_b"}}},
+		Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
+		Fks:     []ddl.Foreignkey{ddl.Foreignkey{Name: "fk1", Columns: []string{"b"}, ReferTable: "ref_table1", ReferColumns: []string{"ref_b"}}},
+		Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "index1", Table: "table1", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}}}},
 	}
 	conv.SpSchema["table2"] = ddl.CreateTable{
 		Name:     "table2",
@@ -62,12 +63,16 @@ func TestGetDDL(t *testing.T) {
 			"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Int64}},
 			"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.Int64}},
 		},
-		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
-		Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk2", Columns: []string{"b", "c"}, ReferTable: "ref_table2", ReferColumns: []string{"ref_b", "ref_c"}}}}
+		Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
+		Fks:     []ddl.Foreignkey{ddl.Foreignkey{Name: "fk2", Columns: []string{"b", "c"}, ReferTable: "ref_table2", ReferColumns: []string{"ref_b", "ref_c"}}},
+		Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "index2", Table: "table2", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: true}, ddl.IndexKey{Col: "c", Desc: false}}}},
+	}
 	tables := conv.GetDDL(ddl.Config{Tables: true, ForeignKeys: false})
 	e := []string{
 		"CREATE TABLE table1 (\n    a INT64,\n    b INT64 \n) PRIMARY KEY (a)",
+		"CREATE INDEX index1 ON table1 (b)",
 		"CREATE TABLE table2 (\n    a INT64,\n    b INT64,\n    c INT64 \n) PRIMARY KEY (a)",
+		"CREATE UNIQUE INDEX index2 ON table2 (b DESC, c)",
 	}
 	assert.ElementsMatch(t, e, tables)
 	fks := conv.GetDDL(ddl.Config{Tables: false, ForeignKeys: true})
@@ -79,7 +84,9 @@ func TestGetDDL(t *testing.T) {
 	tablesAndFks := conv.GetDDL(ddl.Config{Tables: true, ForeignKeys: true})
 	e3 := []string{
 		"CREATE TABLE table1 (\n    a INT64,\n    b INT64 \n) PRIMARY KEY (a)",
+		"CREATE INDEX index1 ON table1 (b)",
 		"CREATE TABLE table2 (\n    a INT64,\n    b INT64,\n    c INT64 \n) PRIMARY KEY (a)",
+		"CREATE UNIQUE INDEX index2 ON table2 (b DESC, c)",
 		"ALTER TABLE table1 ADD CONSTRAINT fk1 FOREIGN KEY (b) REFERENCES ref_table1 (ref_b)",
 		"ALTER TABLE table2 ADD CONSTRAINT fk2 FOREIGN KEY (b, c) REFERENCES ref_table2 (ref_b, ref_c)",
 	}
diff --git a/internal/mapping_test.go b/internal/mapping_test.go
index dc48071..95a2184 100644
--- a/internal/mapping_test.go
+++ b/internal/mapping_test.go
@@ -93,7 +93,7 @@ func TestGetSpannerCol(t *testing.T) {
 	}
 }
 
-func TestGetSpannerKeyName(t *testing.T) {
+func TestToSpannerForeignKey(t *testing.T) {
 	schemaForeignKeys := make(map[string]bool)
 
 	basicTests := []struct {
@@ -103,20 +103,36 @@ func TestGetSpannerKeyName(t *testing.T) {
 	}{
 		{"Good name", "fktest", "fktest"},
 		{"Empty name", "", ""},
-		{"Collision", "fktest", "fktest_1"},
-		{"Collision 2", "fktest", "fktest_2"},
-		{"Bad name", "fk\ntest", "fk_test"},
-		{"Bad name 2", "f\nk\ntest", "f_k_test"},
-		{"Collision 3", "fktest", "fktest_5"},
-		{"Collision 4", "fktest", "fktest_6"},
-		{"Good name", "fk_test_5", "fk_test_5"},
-		{"Collision 5", "fktest_6", "fktest_6_8"},
-		{"Bad name with collision", "fk\ttest", "fk_test_9"},
-		{"Bad name with collision 2", "fk\ntest", "fk_test_10"},
-		{"Bad name with collision 3", "fk?test", "fk_test_11"},
 	}
 	for _, tc := range basicTests {
-		spKeyName := GetSpannerKeyName(tc.srcKeyName, schemaForeignKeys)
+		spKeyName := ToSpannerForeignKey(tc.srcKeyName, schemaForeignKeys)
+		assert.Equal(t, tc.spKeyName, spKeyName, tc.name)
+	}
+}
+
+func TestGetSpannerId(t *testing.T) {
+	schemaIndexKeys := make(map[string]bool)
+
+	basicTests := []struct {
+		name       string // Name of test.
+		srcKeyName string // Source key name.
+		spKeyName  string // Expected Spanner key name.
+	}{
+		{"Good name", "index1", "index1"},
+		{"Collision", "index1", "index1_1"},
+		{"Collision 2", "index1", "index1_2"},
+		{"Bad name", "in\ndex", "in_dex"},
+		{"Bad name 2", "i\nn\ndex", "i_n_dex"},
+		{"Collision 3", "index", "index"},
+		{"Collision 4", "index", "index_6"},
+		{"Good name", "in_dex", "in_dex_7"},
+		{"Collision 5", "index_6", "index_6_8"},
+		{"Bad name with collision", "in\tdex", "in_dex_9"},
+		{"Bad name with collision 2", "in\ndex", "in_dex_10"},
+		{"Bad name with collision 3", "in?dex", "in_dex_11"},
+	}
+	for _, tc := range basicTests {
+		spKeyName := getSpannerId(tc.srcKeyName, schemaIndexKeys)
 		assert.Equal(t, tc.spKeyName, spKeyName, tc.name)
 	}
 }
diff --git a/mysql/infoschema_test.go b/mysql/infoschema_test.go
index 2e26337..ece231d 100644
--- a/mysql/infoschema_test.go
+++ b/mysql/infoschema_test.go
@@ -69,6 +69,11 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"test", "ref", "id", "fk_test"},
 			},
 		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "user"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
+		},
+		{
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"test", "cart"},
 			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
@@ -90,6 +95,16 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			rows: [][]driver.Value{
 				{"product", "productid", "product_id", "fk_test2"},
 				{"user", "userid", "user_id", "fk_test3"}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "cart"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
+			rows: [][]driver.Value{
+				{"index1", "userid", 1, sql.NullString{Valid: false}, "0"},
+				{"index2", "userid", 1, "A", "1"},
+				{"index2", "productid", 2, "D", "1"},
+				{"index3", "productid", 1, "A", "0"},
+				{"index3", "userid", 2, "D", "0"}},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"test", "product"},
@@ -107,6 +122,10 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "product"},
 			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "product"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"test", "test"},
@@ -143,6 +162,10 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
 			rows: [][]driver.Value{{"test_ref", "id", "ref_id", "fk_test4"},
 				{"test_ref", "txt", "ref_txt", "fk_test4"}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"test", "test_ref"},
@@ -162,6 +185,10 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "test_ref"},
 			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "test_ref"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		},
 	}
 	db := mkMockDB(t, ms)
@@ -189,7 +216,10 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			},
 			Pks: []ddl.IndexKey{ddl.IndexKey{Col: "productid"}, ddl.IndexKey{Col: "userid"}},
 			Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test2", Columns: []string{"productid"}, ReferTable: "product", ReferColumns: []string{"product_id"}},
-				ddl.Foreignkey{Name: "fk_test3", Columns: []string{"userid"}, ReferTable: "user", ReferColumns: []string{"user_id"}}}},
+				ddl.Foreignkey{Name: "fk_test3", Columns: []string{"userid"}, ReferTable: "user", ReferColumns: []string{"user_id"}}},
+			Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "index1", Table: "cart", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "userid", Desc: false}}},
+				ddl.CreateIndex{Name: "index2", Table: "cart", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "userid", Desc: false}, ddl.IndexKey{Col: "productid", Desc: true}}},
+				ddl.CreateIndex{Name: "index3", Table: "cart", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "productid", Desc: false}, ddl.IndexKey{Col: "userid", Desc: true}}}}},
 		"product": ddl.CreateTable{
 			Name:     "product",
 			ColNames: []string{"product_id", "product_name"},
@@ -332,6 +362,10 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "test"},
 			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		},
 		// Note: go-sqlmock mocks specify an ordered sequence
 		// of queries and results.  This (repeated) entry is
diff --git a/mysql/mysqldump_test.go b/mysql/mysqldump_test.go
index 75f6068..3549621 100644
--- a/mysql/mysqldump_test.go
+++ b/mysql/mysqldump_test.go
@@ -211,7 +211,7 @@ func TestProcessMySQLDump_MultiCol(t *testing.T) {
 					},
 					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
 					Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "A_fk_test_2", Columns: []string{"d"}, ReferTable: "test", ReferColumns: []string{"a"}},
-						ddl.Foreignkey{Name: "A_fk_test_2_1", Columns: []string{"c"}, ReferTable: "test3", ReferColumns: []string{"e"}}}}},
+						ddl.Foreignkey{Name: "A_fk_test_2_4", Columns: []string{"c"}, ReferTable: "test3", ReferColumns: []string{"e"}}}}},
 		},
 		{
 			name: "Alter table add foreign key",
@@ -325,6 +325,176 @@ func TestProcessMySQLDump_MultiCol(t *testing.T) {
 					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
 					Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test", Columns: []string{"e", "f"}, ReferTable: "test", ReferColumns: []string{"a", "b"}}}}},
 		},
+		{
+			name: "Create table with index keys",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL," +
+				"KEY custom_index (b, c)" +
+				");\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Create table with unique index keys",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL," +
+				"UNIQUE KEY custom_index (b, c)" +
+				");\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Create table with multiple index keys with different order",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL," +
+				"UNIQUE KEY custom_index (b, c)," +
+				"KEY custom_index2 (c, a)" +
+				");\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}},
+						ddl.CreateIndex{Name: "custom_index2", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: false}, ddl.IndexKey{Col: "a", Desc: false}}}}}},
+		},
+		{
+			name: "Alter table add index keys",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL" +
+				");\n" +
+				"ALTER TABLE test ADD INDEX custom_index (b, c);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Alter table add unique index keys",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL" +
+				");\n" +
+				"ALTER TABLE test ADD UNIQUE INDEX custom_index (b, c);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Alter table add index keys",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL" +
+				");\n" +
+				"ALTER TABLE test ADD CONSTRAINT custom_index UNIQUE (b, c);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Create index statement",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL" +
+				");\n" +
+				"CREATE INDEX custom_index ON test (b, c);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Create unique index statement",
+			input: "CREATE TABLE test (" +
+				"a smallint DEFAULT NULL," +
+				"b text DEFAULT NULL," +
+				"c text DEFAULT NULL" +
+				");\n" +
+				"CREATE UNIQUE INDEX custom_index ON test (b, c);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
 		{
 			name:  "Create table with mysql schema",
 			input: "CREATE TABLE myschema.test (a text PRIMARY KEY, b text);\n",
diff --git a/mysql/toddl_test.go b/mysql/toddl_test.go
index b103647..49504de 100644
--- a/mysql/toddl_test.go
+++ b/mysql/toddl_test.go
@@ -42,6 +42,7 @@ func TestToSpannerType(t *testing.T) {
 		PrimaryKeys: []schema.Key{schema.Key{Column: "a"}},
 		ForeignKeys: []schema.ForeignKey{schema.ForeignKey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
 			schema.ForeignKey{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aRef"}}},
+		Indexes: []schema.Index{schema.Index{Name: "index1", Unique: true, Keys: []schema.Key{schema.Key{Column: "a", Desc: false}, schema.Key{Column: "d", Desc: true}}}},
 	}
 	conv.SrcSchema[name] = srcSchema
 	conv.SpSchema["ref_table"] = ddl.CreateTable{
@@ -81,6 +82,7 @@ func TestToSpannerType(t *testing.T) {
 		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
 		Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
 			ddl.Foreignkey{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}},
+		Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "index1", Table: name, Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "a", Desc: false}, ddl.IndexKey{Col: "d", Desc: true}}}},
 	}
 	assert.Equal(t, expected, actual)
 	expectedIssues := map[string][]internal.SchemaIssue{
diff --git a/postgres/infoschema_test.go b/postgres/infoschema_test.go
index a9bf4e2..aea1ce0 100644
--- a/postgres/infoschema_test.go
+++ b/postgres/infoschema_test.go
@@ -69,8 +69,11 @@ func TestProcessInfoSchema(t *testing.T) {
 			rows: [][]driver.Value{
 				{"public", "test", "ref", "id", "fk_test"},
 			},
-		},
-		{
+		}, {
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "user"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"public", "cart"},
 			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
@@ -92,6 +95,16 @@ func TestProcessInfoSchema(t *testing.T) {
 			rows: [][]driver.Value{
 				{"public", "product", "productid", "product_id", "fk_test2"},
 				{"public", "user", "userid", "user_id", "fk_test3"}},
+		}, {
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "cart"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+			rows: [][]driver.Value{{"index1", "userid", 1, "false", "ASC"},
+				{"index2", "userid", 1, "true", "ASC"},
+				{"index2", "productid", 2, "true", "DESC"},
+				{"index3", "productid", 1, "true", "DESC"},
+				{"index3", "userid", 2, "true", "ASC"},
+			},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"public", "product"},
@@ -109,6 +122,10 @@ func TestProcessInfoSchema(t *testing.T) {
 			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
 			args:  []driver.Value{"public", "product"},
 			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+		}, {
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "product"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"public", "test"},
@@ -146,6 +163,10 @@ func TestProcessInfoSchema(t *testing.T) {
 			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
 			rows: [][]driver.Value{{"public", "test_ref", "id", "ref_id", "fk_test4"},
 				{"public", "test_ref", "txt", "ref_txt", "fk_test4"}},
+		}, {
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"public", "test_ref"},
@@ -165,6 +186,10 @@ func TestProcessInfoSchema(t *testing.T) {
 			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
 			args:  []driver.Value{"public", "test_ref"},
 			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+		}, {
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "test_ref"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		},
 	}
 	db := mkMockDB(t, ms)
@@ -192,7 +217,10 @@ func TestProcessInfoSchema(t *testing.T) {
 			},
 			Pks: []ddl.IndexKey{ddl.IndexKey{Col: "productid"}, ddl.IndexKey{Col: "userid"}},
 			Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test2", Columns: []string{"productid"}, ReferTable: "product", ReferColumns: []string{"product_id"}},
-				ddl.Foreignkey{Name: "fk_test3", Columns: []string{"userid"}, ReferTable: "user", ReferColumns: []string{"user_id"}}}},
+				ddl.Foreignkey{Name: "fk_test3", Columns: []string{"userid"}, ReferTable: "user", ReferColumns: []string{"user_id"}}},
+			Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "index1", Table: "cart", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "userid", Desc: false}}},
+				ddl.CreateIndex{Name: "index2", Table: "cart", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "userid", Desc: false}, ddl.IndexKey{Col: "productid", Desc: true}}},
+				ddl.CreateIndex{Name: "index3", Table: "cart", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "productid", Desc: true}, ddl.IndexKey{Col: "userid", Desc: false}}}}},
 		"product": ddl.CreateTable{
 			Name:     "product",
 			ColNames: []string{"product_id", "product_name"},
@@ -411,6 +439,11 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 			args:  []driver.Value{"public", "test"},
 			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
 		},
+		{
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		},
 		// Note: go-sqlmock mocks specify an ordered sequence
 		// of queries and results.  This (repeated) entry is
 		// needed because ProcessSqlData (redundantly) gets
diff --git a/postgres/pgdump_test.go b/postgres/pgdump_test.go
index 746ba23..4d34ecf 100644
--- a/postgres/pgdump_test.go
+++ b/postgres/pgdump_test.go
@@ -272,6 +272,111 @@ func TestProcessPgDump(t *testing.T) {
 					Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test", Columns: []string{"e"}, ReferTable: "test", ReferColumns: []string{"a"}},
 						ddl.Foreignkey{Name: "fk_test2", Columns: []string{"f"}, ReferTable: "test2", ReferColumns: []string{"c"}}}}},
 		},
+		{
+			name: "Create index statement",
+			input: "CREATE TABLE test (" +
+				"a smallint," +
+				"b text," +
+				"c text" +
+				");\n" +
+				"CREATE INDEX custom_index ON test (b, c);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Create index statement with order",
+			input: "CREATE TABLE test (" +
+				"a smallint," +
+				"b text," +
+				"c text" +
+				");\n" +
+				"CREATE INDEX custom_index ON test (b DESC, c ASC);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: true}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
+		{
+			name: "Create index statement with different sequence order",
+			input: "CREATE TABLE test (" +
+				"a smallint," +
+				"b text," +
+				"c text" +
+				");\n" +
+				"CREATE UNIQUE INDEX custom_index ON test (c DESC, b ASC);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "c", Desc: true}, ddl.IndexKey{Col: "b", Desc: false}}}}}},
+		},
+		{
+			name: "Create table with unique constraint",
+			input: "CREATE TABLE test (" +
+				"a smallint," +
+				"b text," +
+				"c text," +
+				"CONSTRAINT custom_index UNIQUE(b)" +
+				");\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}}}}}},
+		},
+		{
+			name: "Alter table add unique constraint",
+			input: "CREATE TABLE test (" +
+				"a smallint," +
+				"b text," +
+				"c text" +
+				");\n" +
+				"ALTER TABLE test ADD CONSTRAINT custom_index UNIQUE (b,c);\n",
+			expectedSchema: map[string]ddl.CreateTable{
+				"test": ddl.CreateTable{
+					Name:     "test",
+					ColNames: []string{"a", "b", "c", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
+						"a":        ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}},
+						"b":        ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"c":        ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+						"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Type{Name: ddl.Int64}},
+					},
+					Pks:     []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}},
+					Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "custom_index", Table: "test", Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}, ddl.IndexKey{Col: "c", Desc: false}}}}}},
+		},
 		{
 			name:  "Create table with pg schema",
 			input: "CREATE TABLE myschema.test (a text PRIMARY KEY, b text);\n",
diff --git a/postgres/toddl_test.go b/postgres/toddl_test.go
index 563cd1b..383e6c0 100644
--- a/postgres/toddl_test.go
+++ b/postgres/toddl_test.go
@@ -44,6 +44,8 @@ func TestToSpannerType(t *testing.T) {
 		PrimaryKeys: []schema.Key{schema.Key{Column: "a"}},
 		ForeignKeys: []schema.ForeignKey{schema.ForeignKey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
 			schema.ForeignKey{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aRef"}}},
+		Indexes: []schema.Index{schema.Index{Name: "index1", Unique: true, Keys: []schema.Key{schema.Key{Column: "a", Desc: false}, schema.Key{Column: "d", Desc: true}}},
+			schema.Index{Name: "index2", Unique: false, Keys: []schema.Key{schema.Key{Column: "d", Desc: true}}}},
 	}
 	conv.SrcSchema[name] = srcSchema
 	conv.SpSchema["ref_table"] = ddl.CreateTable{
@@ -82,7 +84,10 @@ func TestToSpannerType(t *testing.T) {
 		},
 		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
 		Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
-			ddl.Foreignkey{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}}}
+			ddl.Foreignkey{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}},
+		Indexes: []ddl.CreateIndex{ddl.CreateIndex{Name: "index1", Table: name, Unique: true, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "a", Desc: false}, ddl.IndexKey{Col: "d", Desc: true}}},
+			ddl.CreateIndex{Name: "index2", Table: name, Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "d", Desc: true}}}},
+	}
 	assert.Equal(t, expected, actual)
 	expectedIssues := map[string][]internal.SchemaIssue{
 		"b": []internal.SchemaIssue{internal.Widened},
diff --git a/spanner/ddl/ast_test.go b/spanner/ddl/ast_test.go
index 241ac6e..b5cd243 100644
--- a/spanner/ddl/ast_test.go
+++ b/spanner/ddl/ast_test.go
@@ -85,6 +85,7 @@ func TestPrintCreateTable(t *testing.T) {
 		cds,
 		[]IndexKey{IndexKey{Col: "col1", Desc: true}},
 		nil,
+		nil,
 		"",
 	}
 	tests := []struct {
@@ -101,21 +102,31 @@ func TestPrintCreateTable(t *testing.T) {
 }
 
 func TestPrintCreateIndex(t *testing.T) {
-	ci := CreateIndex{
-		"myindex",
-		"mytable",
-		[]IndexKey{IndexKey{Col: "col1", Desc: true}, IndexKey{Col: "col2"}},
-	}
+	ci := []CreateIndex{
+		CreateIndex{
+			"myindex",
+			"mytable",
+			/*Unique =*/ false,
+			[]IndexKey{IndexKey{Col: "col1", Desc: true}, IndexKey{Col: "col2"}},
+		},
+		CreateIndex{
+			"myindex2",
+			"mytable",
+			/*Unique =*/ true,
+			[]IndexKey{IndexKey{Col: "col1", Desc: true}, IndexKey{Col: "col2"}},
+		}}
 	tests := []struct {
 		name       string
 		protectIds bool
+		index      CreateIndex
 		expected   string
 	}{
-		{"no quote", false, "CREATE INDEX myindex ON mytable (col1 DESC, col2)"},
-		{"quote", true, "CREATE INDEX `myindex` ON `mytable` (`col1` DESC, `col2`)"},
+		{"no quote non unique", false, ci[0], "CREATE INDEX myindex ON mytable (col1 DESC, col2)"},
+		{"quote non unique", true, ci[0], "CREATE INDEX `myindex` ON `mytable` (`col1` DESC, `col2`)"},
+		{"unique key", true, ci[1], "CREATE UNIQUE INDEX `myindex2` ON `mytable` (`col1` DESC, `col2`)"},
 	}
 	for _, tc := range tests {
-		assert.Equal(t, normalizeSpace(tc.expected), normalizeSpace(ci.PrintCreateIndex(Config{ProtectIds: tc.protectIds})))
+		assert.Equal(t, normalizeSpace(tc.expected), normalizeSpace(tc.index.PrintCreateIndex(Config{ProtectIds: tc.protectIds})))
 	}
 }