Optional integrator schema validation (#2348)

* Optional integrator schema validation

* [ga-format-pr] Run ./format_repo.sh to fix formatting

* cleanup

* zach comments

* remove bad impl

---------

Co-authored-by: max-hoffman <[email protected]>

Author: max-hoffman

enginetest/engine_only_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/dolthub/vitess/go/sqltypes"
 	"github.com/pmezard/go-difflib/difflib"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -981,6 +982,70 @@ func TestTimestampBindingsCanBeCompared(t *testing.T) {
 	require.Equal(t, 1, count)
 }
 
+// TestAlterTableWithBadSchema is a backwards compatibility test that
+// ensures tables made with old versions of the engine can be altered.
+func TestAlterTableWithBadSchema(t *testing.T) {
+	harness := enginetest.NewDefaultMemoryHarness()
+	pro := harness.Provider()
+	harness.NewDatabases("mydb")
+	ctx := harness.NewContext()
+	sqlDb, err := pro.Database(ctx, "mydb")
+	require.NoError(t, err)
+	db := sqlDb.(*memory.HistoryDatabase)
+
+	sch := sql.NewPrimaryKeySchema(sql.Schema{
+		{Name: "a", Type: types.MustCreateStringWithDefaults(sqltypes.VarChar, 16383), Source: "mytable"},
+		{Name: "b", Type: types.MustCreateStringWithDefaults(sqltypes.VarChar, 16383), Source: "mytable"},
+		{Name: "c", Type: types.MustCreateStringWithDefaults(sqltypes.VarChar, 16383), Source: "mytable"},
+	})
+
+	harness.NewTableAsOf(db, "mytable", sch, nil)
+
+	engine, err := harness.NewEngine(t)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name string
+		q    string
+		err  bool
+	}{
+		{
+			name: "noop modify triggers validation",
+			q:    "alter table mytable rename column a to d",
+			err:  true,
+		},
+		{
+			name: "partial update with invalid final schema fails",
+			q:    "alter table mytable modify column a varchar(100), modify column b varchar(100)",
+			err:  true,
+		},
+		{
+			name: "update with valid final schema succeeds",
+			q:    "alter table mytable modify column a varchar(100), modify column b varchar(100), modify column c varchar(100)",
+			err:  false,
+		},
+		{
+			name: "mixed update add with invalid final schema fails",
+			q:    "alter table mytable modify column a varchar(100), modify column b varchar(100), modify column c varchar(100), add column d varchar(max)",
+			err:  true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := harness.NewContext()
+			_, iter, err := engine.Query(ctx, tt.q)
+			// errors should be analyze time, not execution time
+			if tt.err {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				_, err = sql.RowIterToRows(ctx, iter)
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
 func newDatabase() (*sql2.DB, func()) {
 	// Grab an empty port so that tests do not fail if a specific port is already in use
 	listener, err := net.Listen("tcp", ":0")

enginetest/queries/alter_table_queries.go

@@ -17,13 +17,34 @@ package queries
 import (
 	"github.com/dolthub/vitess/go/mysql"
 
-	"github.com/dolthub/go-mysql-server/sql/plan"
-
 	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/go-mysql-server/sql/analyzer/analyzererrors"
+	"github.com/dolthub/go-mysql-server/sql/plan"
 	"github.com/dolthub/go-mysql-server/sql/types"
 )
 
 var AlterTableScripts = []ScriptTest{
+	{
+		Name: "multi alter with invalid schemas",
+		SetUpScript: []string{
+			"CREATE TABLE t(a int primary key)",
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query:       "alter table t add column b varchar(16383)",
+				ExpectedErr: analyzererrors.ErrInvalidRowLength,
+			},
+			{
+				// 1 char = 4 bytes with default collation
+				Query:       "alter table t add column b varchar(16000), add column c varchar(16000)",
+				ExpectedErr: analyzererrors.ErrInvalidRowLength,
+			},
+			{
+				Query:    "alter table t add column b varchar(16000), add column c varchar(10)",
+				Expected: []sql.Row{{types.NewOkResult(0)}},
+			},
+		},
+	},
 	{
 		Name: "variety of alter column statements in a single statement",
 		SetUpScript: []string{

memory/database.go

@@ -48,6 +48,7 @@ var _ sql.EventDatabase = (*Database)(nil)
 var _ sql.ViewDatabase = (*Database)(nil)
 var _ sql.CollatedDatabase = (*Database)(nil)
 var _ fulltext.Database = (*Database)(nil)
+var _ sql.SchemaValidator = (*BaseDatabase)(nil)
 
 // BaseDatabase is an in-memory database that can't store views, only for testing the engine
 type BaseDatabase struct {
@@ -81,6 +82,11 @@ func NewViewlessDatabase(name string) *BaseDatabase {
 	}
 }
 
+// ValidateSchema implements sql.SchemaValidator
+func (d *BaseDatabase) ValidateSchema(schema sql.Schema) error {
+	return validateMaxRowLength(schema)
+}
+
 // EnablePrimaryKeyIndexes causes every table created in this database to use an index on its primary partitionKeys
 func (d *BaseDatabase) EnablePrimaryKeyIndexes() {
 	d.primaryKeyIndexes = true
@@ -234,10 +240,6 @@ func (d *BaseDatabase) CreateTable(ctx *sql.Context, name string, schema sql.Pri
 		return sql.ErrTableAlreadyExists.New(name)
 	}
 
-	if err := validateMaxRowLength(schema.PhysicalSchema()); err != nil {
-		return err
-	}
-
 	table := NewTableWithCollation(d, name, schema, d.fkColl, collation)
 	table.db = d
 	table.data.comment = comment

memory/table.go

@@ -1165,10 +1165,6 @@ func addColumnToSchema(ctx *sql.Context, data *TableData, newCol *sql.Column, or
 		}
 	}
 
-	if err := validateMaxRowLength(newSch.PhysicalSchema()); err != nil {
-		return 0, nil, err
-	}
-
 	for _, newSchCol := range newSch {
 		newDefault, _, _ := transform.Expr(newSchCol.Default, func(expr sql.Expression) (sql.Expression, transform.TreeIdentity, error) {
 			if expr, ok := expr.(*expression.GetField); ok {

sql/analyzer/validate_create_table.go

@@ -155,6 +155,7 @@ func validateIdentifiers(name string, spec *plan.TableSpec) error {
 func resolveAlterColumn(ctx *sql.Context, a *Analyzer, n sql.Node, scope *plan.Scope, sel RuleSelector) (sql.Node, transform.TreeIdentity, error) {
 	var sch sql.Schema
 	var indexes []string
+	var validator sql.SchemaValidator
 	keyedColumns := make(map[string]bool)
 	var err error
 	transform.Inspect(n, func(n sql.Node) bool {
@@ -163,16 +164,41 @@ func resolveAlterColumn(ctx *sql.Context, a *Analyzer, n sql.Node, scope *plan.S
 		}
 		switch n := n.(type) {
 		case *plan.ModifyColumn:
+			if rt, ok := n.Table.(*plan.ResolvedTable); ok {
+				if sv, ok := rt.UnwrappedDatabase().(sql.SchemaValidator); ok {
+					validator = sv
+				}
+			}
 			keyedColumns, err = getTableIndexColumns(ctx, n.Table)
 			return false
 		case *plan.RenameColumn:
+			if rt, ok := n.Table.(*plan.ResolvedTable); ok {
+				if sv, ok := rt.UnwrappedDatabase().(sql.SchemaValidator); ok {
+					validator = sv
+				}
+			}
 			return false
 		case *plan.AddColumn:
+			if rt, ok := n.Table.(*plan.ResolvedTable); ok {
+				if sv, ok := rt.UnwrappedDatabase().(sql.SchemaValidator); ok {
+					validator = sv
+				}
+			}
 			keyedColumns, err = getTableIndexColumns(ctx, n.Table)
 			return false
 		case *plan.DropColumn:
+			if rt, ok := n.Table.(*plan.ResolvedTable); ok {
+				if sv, ok := rt.UnwrappedDatabase().(sql.SchemaValidator); ok {
+					validator = sv
+				}
+			}
 			return false
 		case *plan.AlterIndex:
+			if rt, ok := n.Table.(*plan.ResolvedTable); ok {
+				if sv, ok := rt.UnwrappedDatabase().(sql.SchemaValidator); ok {
+					validator = sv
+				}
+			}
 			indexes, err = getTableIndexNames(ctx, a, n.Table)
 		default:
 		}
@@ -293,6 +319,12 @@ func resolveAlterColumn(ctx *sql.Context, a *Analyzer, n sql.Node, scope *plan.S
 		return nil, transform.SameTree, err
 	}
 
+	if validator != nil {
+		if err := validator.ValidateSchema(sch); err != nil {
+			return nil, transform.SameTree, err
+		}
+	}
+
 	// We can't evaluate auto-increment until the end of the analysis, since we break adding a new auto-increment unique
 	// column into two steps: first add the column, then create the index. If there was no index created, that's an error.
 	if addedColumn {

sql/plan/resolved_table.go

@@ -18,6 +18,8 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/dolthub/go-mysql-server/sql/mysql_db"
+
 	"github.com/dolthub/go-mysql-server/memory"
 	"github.com/dolthub/go-mysql-server/sql"
 )
@@ -91,6 +93,13 @@ func (t *ResolvedTable) Database() sql.Database {
 	return t.SqlDatabase
 }
 
+func (t *ResolvedTable) UnwrappedDatabase() sql.Database {
+	if privDb, ok := t.SqlDatabase.(mysql_db.PrivilegedDatabase); ok {
+		return privDb.Unwrap()
+	}
+	return t.SqlDatabase
+}
+
 func (t *ResolvedTable) WithDatabase(database sql.Database) (sql.Node, error) {
 	newNode := *t
 	t.SqlDatabase = database

sql/planbuilder/ddl.go

@@ -25,6 +25,7 @@ import (
 	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/dolthub/go-mysql-server/sql/expression"
 	"github.com/dolthub/go-mysql-server/sql/expression/function"
+	"github.com/dolthub/go-mysql-server/sql/mysql_db"
 	"github.com/dolthub/go-mysql-server/sql/plan"
 	"github.com/dolthub/go-mysql-server/sql/types"
 )
@@ -212,35 +213,45 @@ func (b *Builder) buildCreateTable(inScope *scope, c *ast.DDL) (outScope *scope)
 		return b.buildCreateTableLike(inScope, c)
 	}
 
+	qualifier := c.Table.Qualifier.String()
+	if qualifier == "" {
+		qualifier = b.ctx.GetCurrentDatabase()
+	}
+	database := b.resolveDb(qualifier)
+
 	// In the case that no table spec is given but a SELECT Statement return the CREATE TABLE node.
 	// if the table spec != nil it will get parsed below.
 	if c.TableSpec == nil && c.OptSelect != nil {
 		tableSpec := &plan.TableSpec{}
 
 		selectScope := b.buildSelectStmt(inScope, c.OptSelect.Select)
 
-		dbName := strings.ToLower(c.Table.Qualifier.String())
-		if dbName == "" {
-			dbName = b.ctx.GetCurrentDatabase()
-		}
-		db := b.resolveDb(dbName)
-		outScope.node = plan.NewCreateTableSelect(db, c.Table.Name.String(), selectScope.node, tableSpec, plan.IfNotExistsOption(c.IfNotExists), plan.TempTableOption(c.Temporary))
+		outScope.node = plan.NewCreateTableSelect(database, c.Table.Name.String(), selectScope.node, tableSpec, plan.IfNotExistsOption(c.IfNotExists), plan.TempTableOption(c.Temporary))
 		return outScope
 	}
 
 	idxDefs := b.buildIndexDefs(inScope, c.TableSpec)
 
-	qualifier := c.Table.Qualifier.String()
-	if qualifier == "" {
-		qualifier = b.ctx.GetCurrentDatabase()
-	}
-	database := b.resolveDb(qualifier)
 	schema, collation, comment := b.tableSpecToSchema(inScope, outScope, database, strings.ToLower(c.Table.Name.String()), c.TableSpec, false)
 	fkDefs, chDefs := b.buildConstraintsDefs(outScope, c.Table, c.TableSpec)
 
 	schema.Schema = assignColumnIndexesInSchema(schema.Schema)
 	chDefs = assignColumnIndexesInCheckDefs(chDefs, schema.Schema)
 
+	if privDb, ok := database.(mysql_db.PrivilegedDatabase); ok {
+		if sv, ok := privDb.Unwrap().(sql.SchemaValidator); ok {
+			if err := sv.ValidateSchema(schema.PhysicalSchema()); err != nil {
+				b.handleErr(err)
+			}
+		}
+	} else {
+		if sv, ok := database.(sql.SchemaValidator); ok {
+			if err := sv.ValidateSchema(schema.PhysicalSchema()); err != nil {
+				b.handleErr(err)
+			}
+		}
+	}
+
 	tableSpec := &plan.TableSpec{
 		Schema:    schema,
 		IdxDefs:   idxDefs,

sql/tables.go

@@ -393,20 +393,33 @@ type RewritableTable interface {
 	RewriteInserter(ctx *Context, oldSchema, newSchema PrimaryKeySchema, oldColumn, newColumn *Column, idxCols []IndexColumn) (RowInserter, error)
 }
 
-// AlterableTable should be implemented by tables that can receive ALTER TABLE statements to modify their schemas.
+// AlterableTable should be implemented by tables that can receive
+// ALTER TABLE statements to modify their schemas.
 type AlterableTable interface {
 	Table
 	UpdatableTable
 
-	// AddColumn adds a column to this table as given. If non-nil, order specifies where in the schema to add the column.
+	// AddColumn adds a column to this table as given. If non-nil, order
+	// specifies where in the schema to add the column.
 	AddColumn(ctx *Context, column *Column, order *ColumnOrder) error
 	// DropColumn drops the column with the name given.
 	DropColumn(ctx *Context, columnName string) error
-	// ModifyColumn modifies the column with the name given, replacing with the new column definition provided (which may
-	// include a name change). If non-nil, order specifies where in the schema to move the column.
+	// ModifyColumn modifies the column with the name given, replacing
+	// with the new column definition provided (which may include a name
+	// change). If non-nil, order specifies where in the schema to move
+	// the column.
 	ModifyColumn(ctx *Context, columnName string, column *Column, order *ColumnOrder) error
 }
 
+// SchemaValidator is a database that performs schema compatibility checks
+// for CREATE and ALTER TABLE statements.
+type SchemaValidator interface {
+	Database
+	// ValidateSchema lets storage integrators validate whether they can
+	// serialize a given schema.
+	ValidateSchema(Schema) error
+}
+
 // UnresolvedTable is a Table that is either unresolved or deferred for until an asOf resolution.
 // Used by the analyzer during planning, and is not expected to be implemented by integrators.
 type UnresolvedTable interface {