support multi-inherit (#2738)

Author: jycor

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/dolthub/go-icu-regex v0.0.0-20240916130659-0118adc6b662
 	github.com/dolthub/jsonpath v0.0.2-0.20240227200619-19675ab05c71
 	github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81
-	github.com/dolthub/vitess v0.0.0-20241104125316-860772ba6683
+	github.com/dolthub/vitess v0.0.0-20241111235433-a20a5ab9d7c9
 	github.com/go-kit/kit v0.10.0
 	github.com/go-sql-driver/mysql v1.7.2-0.20231213112541-0004702b931d
 	github.com/gocraft/dbr/v2 v2.7.2

go.sum

@@ -58,8 +58,8 @@ github.com/dolthub/jsonpath v0.0.2-0.20240227200619-19675ab05c71 h1:bMGS25NWAGTE
 github.com/dolthub/jsonpath v0.0.2-0.20240227200619-19675ab05c71/go.mod h1:2/2zjLQ/JOOSbbSboojeg+cAwcRV0fDLzIiWch/lhqI=
 github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81 h1:7/v8q9XGFa6q5Ap4Z/OhNkAMBaK5YeuEzwJt+NZdhiE=
 github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81/go.mod h1:siLfyv2c92W1eN/R4QqG/+RjjX5W2+gCTRjZxBjI3TY=
-github.com/dolthub/vitess v0.0.0-20241104125316-860772ba6683 h1:2/RJeUfNAXS7mbBnEr9C36htiCJHk5XldDPzhxtEsME=
-github.com/dolthub/vitess v0.0.0-20241104125316-860772ba6683/go.mod h1:uBvlRluuL+SbEWTCZ68o0xvsdYZER3CEG/35INdzfJM=
+github.com/dolthub/vitess v0.0.0-20241111235433-a20a5ab9d7c9 h1:s36zDuLPuZRWC0nBCJs2Z8joP19eKEtcsIsuE8K9Kx0=
+github.com/dolthub/vitess v0.0.0-20241111235433-a20a5ab9d7c9/go.mod h1:uBvlRluuL+SbEWTCZ68o0xvsdYZER3CEG/35INdzfJM=
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=

sql/planbuilder/ddl.go

@@ -331,85 +331,136 @@ func assignColumnIndexesInSchema(schema sql.Schema) sql.Schema {
 	return newSch
 }
 
-func (b *Builder) buildCreateTableLike(inScope *scope, ct *ast.DDL) *scope {
-	outScope, ok := b.buildTablescan(inScope, ct.OptLike.LikeTable, nil)
-	if !ok {
-		b.handleErr(sql.ErrTableNotFound.New(ct.OptLike.LikeTable.Name.String()))
+func (b *Builder) getIndexDefs(table sql.Table) sql.IndexDefs {
+	idxTbl, isIdxTbl := table.(sql.IndexAddressableTable)
+	if !isIdxTbl {
+		return nil
 	}
-
-	likeTable, ok := outScope.node.(*plan.ResolvedTable)
-	if !ok {
-		err := fmt.Errorf("expected resolved table: %s", ct.OptLike.LikeTable.Name.String())
+	var idxDefs sql.IndexDefs
+	idxs, err := idxTbl.GetIndexes(b.ctx)
+	if err != nil {
 		b.handleErr(err)
 	}
+	for _, idx := range idxs {
+		if idx.IsGenerated() {
+			continue
+		}
+		constraint := sql.IndexConstraint_None
+		if idx.IsUnique() {
+			if idx.ID() == "PRIMARY" {
+				constraint = sql.IndexConstraint_Primary
+			} else {
+				constraint = sql.IndexConstraint_Unique
+			}
+		}
+		columns := make([]sql.IndexColumn, len(idx.Expressions()))
+		for i, col := range idx.Expressions() {
+			// TODO: find a better way to get only the column name if the table is present
+			col = strings.TrimPrefix(col, idxTbl.Name()+".")
+			columns[i] = sql.IndexColumn{Name: col}
+		}
+		idxDefs = append(idxDefs, &sql.IndexDef{
+			Name:       idx.ID(),
+			Storage:    sql.IndexUsing_Default,
+			Constraint: constraint,
+			Columns:    columns,
+			Comment:    idx.Comment(),
+		})
+	}
+	return idxDefs
+}
 
+func (b *Builder) buildCreateTableLike(inScope *scope, ct *ast.DDL) *scope {
+	database := b.resolveDbForTable(ct.Table)
 	newTableName := strings.ToLower(ct.Table.Name.String())
-	outScope.setTableAlias(newTableName)
 
+	var pkSch sql.PrimaryKeySchema
+	var coll sql.CollationID
+	var comment string
+	outScope := inScope.push()
+	if ct.TableSpec != nil {
+		pkSch, coll, _ = b.tableSpecToSchema(inScope, outScope, database, strings.ToLower(ct.Table.Name.String()), ct.TableSpec, false)
+	}
+
+	var ok bool
+	var pkOrdinals []int
+	var newSch sql.Schema
+	newSchMap := make(map[string]struct{})
 	var idxDefs sql.IndexDefs
-	if indexableTable, ok := likeTable.Table.(sql.IndexAddressableTable); ok {
-		indexes, err := indexableTable.GetIndexes(b.ctx)
-		if err != nil {
+	var checkDefs []*sql.CheckConstraint
+	for _, likeTable := range ct.OptLike.LikeTables {
+		outScope, ok = b.buildTablescan(outScope, likeTable, nil)
+		if !ok {
+			b.handleErr(sql.ErrTableNotFound.New(likeTable.Name.String()))
+		}
+		lTable, isResTbl := outScope.node.(*plan.ResolvedTable)
+		if !isResTbl {
+			err := fmt.Errorf("expected resolved table: %s", likeTable.Name.String())
 			b.handleErr(err)
 		}
-		for _, index := range indexes {
-			if index.IsGenerated() {
-				continue
-			}
-			constraint := sql.IndexConstraint_None
-			if index.IsUnique() {
-				if index.ID() == "PRIMARY" {
-					constraint = sql.IndexConstraint_Primary
-				} else {
-					constraint = sql.IndexConstraint_Unique
-				}
-			}
 
-			columns := make([]sql.IndexColumn, len(index.Expressions()))
-			for i, col := range index.Expressions() {
-				//TODO: find a better way to get only the column name if the table is present
-				col = strings.TrimPrefix(col, indexableTable.Name()+".")
-				columns[i] = sql.IndexColumn{Name: col}
-			}
-			idxDefs = append(idxDefs, &sql.IndexDef{
-				Name:       index.ID(),
-				Storage:    sql.IndexUsing_Default,
-				Constraint: constraint,
-				Columns:    columns,
-				Comment:    index.Comment(),
-			})
+		if coll == sql.Collation_Unspecified {
+			coll = lTable.Collation()
 		}
-	}
-	origSch := likeTable.Schema()
-	newSch := make(sql.Schema, len(origSch))
-	for i, col := range origSch {
-		tempCol := *col
-		tempCol.Source = newTableName
-		newSch[i] = &tempCol
-	}
 
-	var pkOrdinals []int
-	if pkTable, ok := likeTable.Table.(sql.PrimaryKeyTable); ok {
-		pkOrdinals = pkTable.PrimaryKeySchema().PkOrdinals
-	}
+		if comment == "" {
+			comment = lTable.Comment()
+		}
 
-	var checkDefs []*sql.CheckConstraint
-	if checksTable, ok := likeTable.Table.(sql.CheckTable); ok {
-		checks, err := checksTable.GetChecks(b.ctx)
-		if err != nil {
-			b.handleErr(err)
+		schOff := len(newSch)
+		hasSkippedCols := false
+		for _, col := range lTable.Schema() {
+			newCol := *col
+			name := strings.ToLower(newCol.Name)
+			if _, ok := newSchMap[name]; ok {
+				// TODO: throw warning
+				hasSkippedCols = true
+				continue
+			}
+			newSchMap[name] = struct{}{}
+			newCol.Source = newTableName
+			newSch = append(newSch, &newCol)
 		}
 
-		for _, check := range checks {
-			checkConstraint := b.buildCheckConstraint(outScope, &check)
-			if err != nil {
-				b.handleErr(err)
+		// if a column was skipped due to duplicates, don't copy over PK ords, idxDefs, or checkDefs
+		// since they might be incorrect
+		if hasSkippedCols {
+			continue
+		}
+
+		// Copy over primary key schema ordinals
+		if pkTable, isPkTable := lTable.Table.(sql.PrimaryKeyTable); isPkTable {
+			for _, pkOrd := range pkTable.PrimaryKeySchema().PkOrdinals {
+				pkOrdinals = append(pkOrdinals, schOff+pkOrd)
 			}
+		}
+
+		// Load index definitions
+		idxDefs = append(idxDefs, b.getIndexDefs(lTable.Table)...)
 
+		// Load check constraints
+		newCheckDefs := b.loadChecksFromTable(outScope, lTable.Table)
+		for _, check := range newCheckDefs {
 			// Prevent a name collision between old and new checks.
-			// New check will be assigned a name during building.
-			checkConstraint.Name = ""
-			checkDefs = append(checkDefs, checkConstraint)
+			// New check name will be assigned a name during building.
+			check.Name = ""
+		}
+		checkDefs = append(checkDefs, newCheckDefs...)
+	}
+
+	var hasSkippedCols bool
+	for _, col := range pkSch.Schema {
+		name := strings.ToLower(col.Name)
+		if _, ok := newSchMap[name]; ok {
+			// TODO: throw warning
+			hasSkippedCols = true
+			continue
+		}
+		newSch = append(newSch, col)
+	}
+	if !hasSkippedCols {
+		for _, pkOrd := range pkSch.PkOrdinals {
+			pkOrdinals = append(pkOrdinals, len(newSch)+pkOrd)
 		}
 	}
 
@@ -420,13 +471,13 @@ func (b *Builder) buildCreateTableLike(inScope *scope, ct *ast.DDL) *scope {
 		Schema:    pkSchema,
 		IdxDefs:   idxDefs,
 		ChDefs:    checkDefs,
-		Collation: likeTable.Collation(),
-		Comment:   likeTable.Comment(),
+		Collation: coll,
+		Comment:   comment,
 	}
 
-	database := b.resolveDbForTable(ct.Table)
-
 	b.qFlags.Set(sql.QFlagSetDatabase)
+
+	outScope.setTableAlias(newTableName)
 	outScope.node = plan.NewCreateTable(database, newTableName, ct.IfNotExists, ct.Temporary, tableSpec)
 	return outScope
 }