// 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 infoschema_test

import (
	"sync"
	"testing"

	. "github.com/pingcap/check"
	"github.com/pingcap/errors"
	"github.com/pingcap/parser/model"
	"github.com/pingcap/parser/mysql"
	"github.com/pingcap/tidb/infoschema"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/meta"
	"github.com/pingcap/tidb/session"
	"github.com/pingcap/tidb/store/mockstore"
	"github.com/pingcap/tidb/types"
	"github.com/pingcap/tidb/util"
	"github.com/pingcap/tidb/util/testleak"
	"github.com/pingcap/tidb/util/testutil"
)

func TestT(t *testing.T) {
	CustomVerboseFlag = true
	TestingT(t)
}

var _ = Suite(&testSuite{})

type testSuite struct {
}

func (*testSuite) TestT(c *C) {
	defer testleak.AfterTest(c)()
	store, err := mockstore.NewMockTikvStore()
	c.Assert(err, IsNil)
	defer store.Close()
	// Make sure it calls perfschema.Init().
	dom, err := session.BootstrapSession(store)
	c.Assert(err, IsNil)
	defer dom.Close()

	handle := infoschema.NewHandle(store)
	dbName := model.NewCIStr("Test")
	tbName := model.NewCIStr("T")
	colName := model.NewCIStr("A")
	idxName := model.NewCIStr("idx")
	noexist := model.NewCIStr("noexist")

	colID, err := genGlobalID(store)
	c.Assert(err, IsNil)
	colInfo := &model.ColumnInfo{
		ID:        colID,
		Name:      colName,
		Offset:    0,
		FieldType: *types.NewFieldType(mysql.TypeLonglong),
		State:     model.StatePublic,
	}

	idxInfo := &model.IndexInfo{
		Name:  idxName,
		Table: tbName,
		Columns: []*model.IndexColumn{
			{
				Name:   colName,
				Offset: 0,
				Length: 10,
			},
		},
		Unique:  true,
		Primary: true,
		State:   model.StatePublic,
	}

	tbID, err := genGlobalID(store)
	c.Assert(err, IsNil)
	tblInfo := &model.TableInfo{
		ID:      tbID,
		Name:    tbName,
		Columns: []*model.ColumnInfo{colInfo},
		Indices: []*model.IndexInfo{idxInfo},
		State:   model.StatePublic,
	}

	dbID, err := genGlobalID(store)
	c.Assert(err, IsNil)
	dbInfo := &model.DBInfo{
		ID:     dbID,
		Name:   dbName,
		Tables: []*model.TableInfo{tblInfo},
		State:  model.StatePublic,
	}

	dbInfos := []*model.DBInfo{dbInfo}
	err = kv.RunInNewTxn(store, true, func(txn kv.Transaction) error {
		meta.NewMeta(txn).CreateDatabase(dbInfo)
		return errors.Trace(err)
	})
	c.Assert(err, IsNil)

	builder, err := infoschema.NewBuilder(handle).InitWithDBInfos(dbInfos, 1)
	c.Assert(err, IsNil)

	txn, err := store.Begin()
	c.Assert(err, IsNil)
	checkApplyCreateNonExistsSchemaDoesNotPanic(c, txn, builder)
	checkApplyCreateNonExistsTableDoesNotPanic(c, txn, builder, dbID)
	txn.Rollback()

	builder.Build()
	is := handle.Get()

	schemaNames := is.AllSchemaNames()
	c.Assert(schemaNames, HasLen, 4)
	c.Assert(testutil.CompareUnorderedStringSlice(schemaNames, []string{util.InformationSchemaName.O, util.MetricSchemaName.O, util.PerformanceSchemaName.O, "Test"}), IsTrue)

	schemas := is.AllSchemas()
	c.Assert(schemas, HasLen, 4)
	schemas = is.Clone()
	c.Assert(schemas, HasLen, 4)

	c.Assert(is.SchemaExists(dbName), IsTrue)
	c.Assert(is.SchemaExists(noexist), IsFalse)

	schema, ok := is.SchemaByID(dbID)
	c.Assert(ok, IsTrue)
	c.Assert(schema, NotNil)

	schema, ok = is.SchemaByID(tbID)
	c.Assert(ok, IsFalse)
	c.Assert(schema, IsNil)

	schema, ok = is.SchemaByName(dbName)
	c.Assert(ok, IsTrue)
	c.Assert(schema, NotNil)

	schema, ok = is.SchemaByName(noexist)
	c.Assert(ok, IsFalse)
	c.Assert(schema, IsNil)

	schema, ok = is.SchemaByTable(tblInfo)
	c.Assert(ok, IsTrue)
	c.Assert(schema, NotNil)

	noexistTblInfo := &model.TableInfo{ID: 12345, Name: tblInfo.Name}
	schema, ok = is.SchemaByTable(noexistTblInfo)
	c.Assert(ok, IsFalse)
	c.Assert(schema, IsNil)

	c.Assert(is.TableExists(dbName, tbName), IsTrue)
	c.Assert(is.TableExists(dbName, noexist), IsFalse)
	c.Assert(is.TableIsView(dbName, tbName), IsFalse)
	c.Assert(is.TableIsSequence(dbName, tbName), IsFalse)

	tb, ok := is.TableByID(tbID)
	c.Assert(ok, IsTrue)
	c.Assert(tb, NotNil)

	tb, ok = is.TableByID(dbID)
	c.Assert(ok, IsFalse)
	c.Assert(tb, IsNil)

	alloc, ok := is.AllocByID(tbID)
	c.Assert(ok, IsTrue)
	c.Assert(alloc, NotNil)

	tb, err = is.TableByName(dbName, tbName)
	c.Assert(err, IsNil)
	c.Assert(tb, NotNil)

	_, err = is.TableByName(dbName, noexist)
	c.Assert(err, NotNil)

	tbs := is.SchemaTables(dbName)
	c.Assert(tbs, HasLen, 1)

	tbs = is.SchemaTables(noexist)
	c.Assert(tbs, HasLen, 0)

	// Make sure partitions table exists
	tb, err = is.TableByName(model.NewCIStr("information_schema"), model.NewCIStr("partitions"))
	c.Assert(err, IsNil)
	c.Assert(tb, NotNil)

	err = kv.RunInNewTxn(store, true, func(txn kv.Transaction) error {
		meta.NewMeta(txn).CreateTableOrView(dbID, tblInfo)
		return errors.Trace(err)
	})
	c.Assert(err, IsNil)
	txn, err = store.Begin()
	c.Assert(err, IsNil)
	_, err = builder.ApplyDiff(meta.NewMeta(txn), &model.SchemaDiff{Type: model.ActionRenameTable, SchemaID: dbID, TableID: tbID, OldSchemaID: dbID})
	c.Assert(err, IsNil)
	txn.Rollback()
	builder.Build()
	is = handle.Get()
	schema, ok = is.SchemaByID(dbID)
	c.Assert(ok, IsTrue)
	c.Assert(len(schema.Tables), Equals, 1)

	emptyHandle := handle.EmptyClone()
	c.Assert(emptyHandle.Get(), IsNil)
}

func (testSuite) TestMockInfoSchema(c *C) {
	tblID := int64(1234)
	tblName := model.NewCIStr("tbl_m")
	tableInfo := &model.TableInfo{
		ID:    tblID,
		Name:  tblName,
		State: model.StatePublic,
	}
	colInfo := &model.ColumnInfo{
		State:     model.StatePublic,
		Offset:    0,
		Name:      model.NewCIStr("h"),
		FieldType: *types.NewFieldType(mysql.TypeLong),
		ID:        1,
	}
	tableInfo.Columns = []*model.ColumnInfo{colInfo}
	is := infoschema.MockInfoSchema([]*model.TableInfo{tableInfo})
	tbl, ok := is.TableByID(tblID)
	c.Assert(ok, IsTrue)
	c.Assert(tbl.Meta().Name, Equals, tblName)
	c.Assert(tbl.Cols()[0].ColumnInfo, Equals, colInfo)
}

func checkApplyCreateNonExistsSchemaDoesNotPanic(c *C, txn kv.Transaction, builder *infoschema.Builder) {
	m := meta.NewMeta(txn)
	_, err := builder.ApplyDiff(m, &model.SchemaDiff{Type: model.ActionCreateSchema, SchemaID: 999})
	c.Assert(infoschema.ErrDatabaseNotExists.Equal(err), IsTrue)
}

func checkApplyCreateNonExistsTableDoesNotPanic(c *C, txn kv.Transaction, builder *infoschema.Builder, dbID int64) {
	m := meta.NewMeta(txn)
	_, err := builder.ApplyDiff(m, &model.SchemaDiff{Type: model.ActionCreateTable, SchemaID: dbID, TableID: 999})
	c.Assert(infoschema.ErrTableNotExists.Equal(err), IsTrue)
}

// TestConcurrent makes sure it is safe to concurrently create handle on multiple stores.
func (testSuite) TestConcurrent(c *C) {
	defer testleak.AfterTest(c)()
	storeCount := 5
	stores := make([]kv.Storage, storeCount)
	for i := 0; i < storeCount; i++ {
		store, err := mockstore.NewMockTikvStore()
		c.Assert(err, IsNil)
		stores[i] = store
	}
	defer func() {
		for _, store := range stores {
			store.Close()
		}
	}()
	var wg sync.WaitGroup
	wg.Add(storeCount)
	for _, store := range stores {
		go func(s kv.Storage) {
			defer wg.Done()
			_ = infoschema.NewHandle(s)
		}(store)
	}
	wg.Wait()
}

// TestInfoTables makes sure that all tables of information_schema could be found in infoschema handle.
func (*testSuite) TestInfoTables(c *C) {
	defer testleak.AfterTest(c)()
	store, err := mockstore.NewMockTikvStore()
	c.Assert(err, IsNil)
	defer store.Close()
	handle := infoschema.NewHandle(store)
	builder, err := infoschema.NewBuilder(handle).InitWithDBInfos(nil, 0)
	c.Assert(err, IsNil)
	builder.Build()
	is := handle.Get()
	c.Assert(is, NotNil)

	infoTables := []string{
		"SCHEMATA",
		"TABLES",
		"COLUMNS",
		"STATISTICS",
		"CHARACTER_SETS",
		"COLLATIONS",
		"FILES",
		"PROFILING",
		"PARTITIONS",
		"KEY_COLUMN_USAGE",
		"REFERENTIAL_CONSTRAINTS",
		"SESSION_VARIABLES",
		"PLUGINS",
		"TABLE_CONSTRAINTS",
		"TRIGGERS",
		"USER_PRIVILEGES",
		"ENGINES",
		"VIEWS",
		"ROUTINES",
		"SCHEMA_PRIVILEGES",
		"COLUMN_PRIVILEGES",
		"TABLE_PRIVILEGES",
		"PARAMETERS",
		"EVENTS",
		"GLOBAL_STATUS",
		"GLOBAL_VARIABLES",
		"SESSION_STATUS",
		"OPTIMIZER_TRACE",
		"TABLESPACES",
		"COLLATION_CHARACTER_SET_APPLICABILITY",
		"PROCESSLIST",
	}
	for _, t := range infoTables {
		tb, err1 := is.TableByName(util.InformationSchemaName, model.NewCIStr(t))
		c.Assert(err1, IsNil)
		c.Assert(tb, NotNil)
	}
}

func genGlobalID(store kv.Storage) (int64, error) {
	var globalID int64
	err := kv.RunInNewTxn(store, true, func(txn kv.Transaction) error {
		var err error
		globalID, err = meta.NewMeta(txn).GenGlobalID()
		return errors.Trace(err)
	})
	return globalID, errors.Trace(err)
}