Ftr: create index (arana-db#254)

* feat: create index

* fix: add space behind create index restore

* fix: create index add index part spec and tests

* fix: remove unnecessary create index tests

* fix: add create index logic table check

pkg/executor/redirect.go

@@ -261,7 +261,7 @@ func (executor *RedirectExecutor) ExecutorComQuery(ctx *proto.Context) (proto.Re
 		} else {
 			err = errNoDatabaseSelected
 		}
-	case *ast.TruncateTableStmt, *ast.DropTableStmt, *ast.ExplainStmt, *ast.DropIndexStmt:
+	case *ast.TruncateTableStmt, *ast.DropTableStmt, *ast.ExplainStmt, *ast.DropIndexStmt, *ast.CreateIndexStmt:
 		res, warn, err = executeStmt(ctx, schemaless, rt)
 	case *ast.DropTriggerStmt:
 		res, warn, err = rt.Execute(ctx)

pkg/runtime/ast/ast.go

@@ -112,6 +112,8 @@ func FromStmtNode(node ast.StmtNode) (Statement, error) {
 		return cc.convDropIndexStmt(stmt), nil
 	case *ast.DropTriggerStmt:
 		return cc.convDropTrigger(stmt), nil
+	case *ast.CreateIndexStmt:
+		return cc.convCreateIndexStmt(stmt), nil
 	default:
 		return nil, errors.Errorf("unimplement: stmt type %T!", stmt)
 	}
@@ -130,6 +132,28 @@ func (cc *convCtx) convDropIndexStmt(stmt *ast.DropIndexStmt) *DropIndexStatemen
 	}
 }
 
+func (cc *convCtx) convCreateIndexStmt(stmt *ast.CreateIndexStmt) *CreateIndexStatement {
+	var tableName TableName
+	if db := stmt.Table.Schema.O; len(db) > 0 {
+		tableName = append(tableName, db)
+	}
+	tableName = append(tableName, stmt.Table.Name.O)
+
+	keys := make([]*IndexPartSpec, len(stmt.IndexPartSpecifications))
+	for i, k := range stmt.IndexPartSpecifications {
+		keys[i] = &IndexPartSpec{
+			Column: cc.convColumn(k.Column),
+			Expr:   toExpressionNode(cc.convExpr(k.Expr)),
+		}
+	}
+
+	return &CreateIndexStatement{
+		Table:     tableName,
+		IndexName: stmt.IndexName,
+		Keys:      keys,
+	}
+}
+
 func (cc *convCtx) convAlterTableStmt(stmt *ast.AlterTableStmt) *AlterTableStatement {
 	var tableName TableName
 	if db := stmt.Table.Schema.O; len(db) > 0 {

pkg/runtime/ast/create_index.go

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ast
+
+import "strings"
+
+var (
+	_ Statement = (*CreateIndexStatement)(nil)
+	_ Restorer  = (*CreateIndexStatement)(nil)
+)
+
+type CreateIndexStatement struct {
+	IndexName string
+	Table     TableName
+	Keys      []*IndexPartSpec
+}
+
+func (c *CreateIndexStatement) CntParams() int {
+	return 0
+}
+
+func (c *CreateIndexStatement) Restore(flag RestoreFlag, sb *strings.Builder, args *[]int) error {
+	sb.WriteString("CREATE INDEX ")
+	sb.WriteString(c.IndexName)
+	if len(c.Table) == 0 {
+		return nil
+	}
+	sb.WriteString(" ON ")
+	if err := c.Table.Restore(flag, sb, args); err != nil {
+		return err
+	}
+
+	sb.WriteString(" (")
+	for i, k := range c.Keys {
+		if i != 0 {
+			sb.WriteString(", ")
+		}
+		if err := k.Restore(flag, sb, args); err != nil {
+			return err
+		}
+	}
+	sb.WriteString(")")
+	return nil
+}
+
+func (c *CreateIndexStatement) Validate() error {
+	return nil
+}
+
+func (c *CreateIndexStatement) Mode() SQLType {
+	return CreateIndex
+}

pkg/runtime/ast/proto.go

@@ -32,6 +32,7 @@ const (
 	SdropTable          // DROP TABLE
 	SalterTable         // ALTER TABLE
 	DropIndex           // DROP INDEX
+	CreateIndex         // CREATE INDEX
 	DropTrigger         // DROP TRIGGER
 )
 
@@ -56,6 +57,7 @@ var _sqlTypeNames = [...]string{
 	SdropTable:  "DROP TABLE",
 	SalterTable: "ALTER TABLE",
 	DropIndex:   "DROP INDEX",
+	CreateIndex: "CREATE INDEX",
 }
 
 // SQLType represents the type of SQL.

pkg/runtime/optimize/optimizer.go

@@ -141,6 +141,8 @@ func (o optimizer) doOptimize(ctx context.Context, conn proto.VConn, stmt rast.S
 		return o.optimizeAlterTable(ctx, t, args)
 	case *rast.DropIndexStatement:
 		return o.optimizeDropIndex(ctx, t, args)
+	case *rast.CreateIndexStatement:
+		return o.optimizeCreateIndex(ctx, t, args)
 	case *rast.DropTriggerStatement:
 		return o.optimizeTrigger(ctx, t, args)
 	}
@@ -172,6 +174,33 @@ func (o optimizer) optimizeDropIndex(ctx context.Context, stmt *rast.DropIndexSt
 	return shardPlan, nil
 }
 
+func (o optimizer) optimizeCreateIndex(ctx context.Context, stmt *rast.CreateIndexStatement, args []interface{}) (proto.Plan, error) {
+
+	var (
+		ret = plan.NewCreateIndexPlan(stmt)
+		ru  = rcontext.Rule(ctx)
+		vt  *rule.VTable
+	)
+
+	//table shard
+	_, ok := ru.VTable(stmt.Table.String())
+	if !ok {
+		return ret, nil
+	}
+
+	// sharding
+	shards := rule.DatabaseTables{}
+	topology := vt.Topology()
+	topology.Each(func(dbIdx, tbIdx int) bool {
+		if d, t, ok := topology.Render(dbIdx, tbIdx); ok {
+			shards[d] = append(shards[d], t)
+		}
+		return true
+	})
+	ret.SetShard(shards)
+	return ret, nil
+}
+
 func (o optimizer) optimizeAlterTable(ctx context.Context, stmt *rast.AlterTableStatement, args []interface{}) (proto.Plan, error) {
 	var (
 		ret   = plan.NewAlterTablePlan(stmt)

pkg/runtime/plan/create_index.go

@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package plan
+
+import (
+	"context"
+	"strings"
+)
+
+import (
+	"github.com/pkg/errors"
+)
+
+import (
+	"github.com/arana-db/arana/pkg/proto"
+	"github.com/arana-db/arana/pkg/proto/rule"
+	"github.com/arana-db/arana/pkg/resultx"
+	"github.com/arana-db/arana/pkg/runtime/ast"
+)
+
+type CreateIndexPlan struct {
+	basePlan
+	stmt   *ast.CreateIndexStatement
+	Shards rule.DatabaseTables
+}
+
+func NewCreateIndexPlan(stmt *ast.CreateIndexStatement) *CreateIndexPlan {
+	return &CreateIndexPlan{
+		stmt: stmt,
+	}
+}
+
+func (c *CreateIndexPlan) Type() proto.PlanType {
+	return proto.PlanTypeExec
+}
+
+func (c *CreateIndexPlan) ExecIn(ctx context.Context, conn proto.VConn) (proto.Result, error) {
+	var (
+		sb   strings.Builder
+		args []int
+	)
+
+	for db, tables := range c.Shards {
+		for i := range tables {
+			table := tables[i]
+
+			stmt := new(ast.CreateIndexStatement)
+			stmt.Table = ast.TableName{table}
+			stmt.IndexName = c.stmt.IndexName
+			stmt.Keys = c.stmt.Keys
+
+			if err := stmt.Restore(ast.RestoreDefault, &sb, &args); err != nil {
+				return nil, err
+			}
+
+			if err := c.execOne(ctx, conn, db, sb.String(), c.toArgs(args)); err != nil {
+				return nil, errors.WithStack(err)
+			}
+
+			sb.Reset()
+		}
+	}
+	return resultx.New(), nil
+}
+
+func (c *CreateIndexPlan) SetShard(shard rule.DatabaseTables) {
+	c.Shards = shard
+}
+
+func (c *CreateIndexPlan) execOne(ctx context.Context, conn proto.VConn, db, query string, args []interface{}) error {
+	res, err := conn.Exec(ctx, db, query, args...)
+	if err != nil {
+		return err
+	}
+	_, _ = res.Dataset()
+	return nil
+}

pkg/runtime/plan/transparent.go

@@ -46,7 +46,7 @@ func Transparent(stmt rast.Statement, args []interface{}) *TransparentPlan {
 	var typ proto.PlanType
 	switch stmt.Mode() {
 	case rast.Sinsert, rast.Sdelete, rast.Sreplace, rast.Supdate, rast.Struncate, rast.SdropTable,
-		rast.SalterTable, rast.DropIndex:
+		rast.SalterTable, rast.DropIndex, rast.CreateIndex:
 		typ = proto.PlanTypeExec
 	default:
 		typ = proto.PlanTypeQuery

test/integration_test.go

@@ -498,6 +498,19 @@ func (s *IntegrationSuite) TestAlterTable() {
 	assert.Equal(t, int64(0), affected)
 }
 
+func (s *IntegrationSuite) TestCreateIndex() {
+	var (
+		db = s.DB()
+		t  = s.T()
+	)
+
+	result, err := db.Exec("create index `name` on student (name)")
+	assert.NoErrorf(t, err, "create index error: %v", err)
+	affected, err := result.RowsAffected()
+	assert.NoErrorf(t, err, "create index error: %v", err)
+	assert.Equal(t, int64(0), affected)
+}
+
 func (s *IntegrationSuite) TestDropIndex() {
 	var (
 		db = s.DB()