You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

218 lines
5.5 KiB

// Copyright 2015 PingCAP, Inc.
//
// Licensed 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,
// See the License for the specific language governing permissions and
// limitations under the License.
package ddltest
import (
"fmt"
"io"
"math"
"sync"
"sync/atomic"
"time"
. "github.com/pingcap/check"
"github.com/pingcap/parser/model"
"github.com/pingcap/parser/terror"
"github.com/pingcap/tidb/store/tikv"
"github.com/pingcap/tidb/store/tikv/gcworker"
"github.com/pingcap/tidb/table"
"github.com/pingcap/tidb/table/tables"
"github.com/pingcap/tidb/types"
goctx "golang.org/x/net/context"
)
func getIndex(t table.Table, name string) table.Index {
for _, idx := range t.Indices() {
if idx.Meta().Name.O == name {
return idx
}
}
return nil
}
func (s *TestDDLSuite) checkAddIndex(c *C, indexInfo *model.IndexInfo) {
ctx := s.ctx
err := ctx.NewTxn(goctx.Background())
c.Assert(err, IsNil)
tbl := s.getTable(c, "test_index")
// read handles form table
handles := make(map[int64]struct{})
err = tbl.IterRecords(ctx, tbl.FirstKey(), tbl.Cols(),
func(h int64, data []types.Datum, cols []*table.Column) (bool, error) {
handles[h] = struct{}{}
return true, nil
})
c.Assert(err, IsNil)
// read handles from index
idx := tables.NewIndex(tbl.Meta().ID, tbl.Meta(), indexInfo)
err = ctx.NewTxn(goctx.Background())
c.Assert(err, IsNil)
txn, err := ctx.Txn(false)
c.Assert(err, IsNil)
defer func() {
txn.Rollback()
}()
it, err := idx.SeekFirst(txn)
c.Assert(err, IsNil)
defer it.Close()
for {
_, h, err := it.Next()
if terror.ErrorEqual(err, io.EOF) {
break
}
c.Assert(err, IsNil)
c.Assert(handles, HasKey, h)
delete(handles, h)
}
c.Assert(handles, HasLen, 0)
}
func (s *TestDDLSuite) checkDropIndex(c *C, indexInfo *model.IndexInfo) {
gcWorker, err := gcworker.NewMockGCWorker(s.store.(tikv.Storage))
c.Assert(err, IsNil)
err = gcWorker.DeleteRanges(goctx.Background(), uint64(math.MaxInt32))
c.Assert(err, IsNil)
ctx := s.ctx
err = ctx.NewTxn(goctx.Background())
c.Assert(err, IsNil)
tbl := s.getTable(c, "test_index")
// read handles from index
idx := tables.NewIndex(tbl.Meta().ID, tbl.Meta(), indexInfo)
err = ctx.NewTxn(goctx.Background())
c.Assert(err, IsNil)
txn, err := ctx.Txn(false)
c.Assert(err, IsNil)
defer txn.Rollback()
it, err := idx.SeekFirst(txn)
c.Assert(err, IsNil)
defer it.Close()
handles := make(map[int64]struct{})
for {
_, h, err := it.Next()
if terror.ErrorEqual(err, io.EOF) {
break
}
c.Assert(err, IsNil)
handles[h] = struct{}{}
}
// TODO: Uncomment this after apply pool is finished
// c.Assert(handles, HasLen, 0)
}
// TestIndex operations on table test_index (c int, c1 bigint, c2 double, c3 varchar(256), primary key(c)).
func (s *TestDDLSuite) TestIndex(c *C) {
// first add many data
workerNum := 10
base := *dataNum / workerNum
var wg sync.WaitGroup
wg.Add(workerNum)
for i := 0; i < workerNum; i++ {
go func(i int) {
defer wg.Done()
for j := 0; j < base; j++ {
k := base*i + j
s.execInsert(c,
fmt.Sprintf("insert into test_index values (%d, %d, %f, '%s')",
k, randomInt(), randomFloat(), randomString(10)))
}
}(i)
}
wg.Wait()
tbl := []struct {
Query string
IndexName string
Add bool
}{
{"create index c1_index on test_index (c1)", "c1_index", true},
{"drop index c1_index on test_index", "c1_index", false},
{"create index c2_index on test_index (c2)", "c2_index", true},
{"drop index c2_index on test_index", "c2_index", false},
{"create index c3_index on test_index (c3)", "c3_index", true},
{"drop index c3_index on test_index", "c3_index", false},
}
insertID := int64(*dataNum)
var oldIndex table.Index
for _, t := range tbl {
c.Logf("run DDL sql %s", t.Query)
done := s.runDDL(t.Query)
ticker := time.NewTicker(time.Duration(*lease) * time.Second / 2)
defer ticker.Stop()
LOOP:
for {
select {
case err := <-done:
c.Assert(err, IsNil)
break LOOP
case <-ticker.C:
// add count new data
// delete count old data randomly
// update count old data randomly
count := 10
s.execIndexOperations(c, workerNum, count, &insertID)
}
}
tbl := s.getTable(c, "test_index")
index := getIndex(tbl, t.IndexName)
if t.Add {
c.Assert(index, NotNil)
oldIndex = index
s.checkAddIndex(c, index.Meta())
} else {
c.Assert(index, IsNil)
s.checkDropIndex(c, oldIndex.Meta())
}
}
}
func (s *TestDDLSuite) execIndexOperations(c *C, workerNum, count int, insertID *int64) {
var wg sync.WaitGroup
// workerNum = 10
wg.Add(workerNum)
for i := 0; i < workerNum; i++ {
go func() {
defer wg.Done()
for j := 0; j < count; j++ {
id := atomic.AddInt64(insertID, 1)
sql := fmt.Sprintf("insert into test_index values (%d, %d, %f, '%s')", id, randomInt(), randomFloat(), randomString(10))
s.execInsert(c, sql)
c.Logf("sql %s", sql)
sql = fmt.Sprintf("delete from test_index where c = %d", randomIntn(int(id)))
s.mustExec(c, sql)
c.Logf("sql %s", sql)
sql = fmt.Sprintf("update test_index set c1 = %d, c2 = %f, c3 = '%s' where c = %d", randomInt(), randomFloat(), randomString(10), randomIntn(int(id)))
s.mustExec(c, sql)
c.Logf("sql %s", sql)
}
}()
}
wg.Wait()
}