// Copyright 2019 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,
// 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 executor_test

import (
	"context"
	"sync"
	"testing"
	"time"

	"github.com/pingcap/failpoint"
	rmpb "github.com/pingcap/kvproto/pkg/resource_manager"
	"github.com/pingcap/tidb/pkg/config"
	distsqlctx "github.com/pingcap/tidb/pkg/distsql/context"
	"github.com/pingcap/tidb/pkg/executor"
	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/parser/auth"
	"github.com/pingcap/tidb/pkg/resourcegroup"
	"github.com/pingcap/tidb/pkg/session"
	"github.com/pingcap/tidb/pkg/sessionctx/variable"
	"github.com/pingcap/tidb/pkg/testkit"
	"github.com/pingcap/tidb/pkg/util/execdetails"
	"github.com/pingcap/tidb/pkg/util/mock"
	"github.com/stretchr/testify/require"
	"github.com/tikv/client-go/v2/util"
)

type mockRUV2ConsumptionReporter struct {
	group     string
	tikvRUV2  float64
	tidbRUV2  float64
	tiflashRU float64
}

func (*mockRUV2ConsumptionReporter) ReportConsumption(_ string, _ *rmpb.Consumption) {}

func (m *mockRUV2ConsumptionReporter) ReportRUV2Consumption(resourceGroupName string, tikvRUV2, tidbRUV2, tiflashRUV2 float64) {
	m.group = resourceGroupName
	m.tikvRUV2 = tikvRUV2
	m.tidbRUV2 = tidbRUV2
	m.tiflashRU = tiflashRUV2
}

type mockRUV2ReportingContext struct {
	*mock.Context
	reporter resourcegroup.ConsumptionReporter
}

func (c *mockRUV2ReportingContext) GetDistSQLCtx() *distsqlctx.DistSQLContext {
	dctx := c.Context.GetDistSQLCtx()
	dctx.RUConsumptionReporter = c.reporter
	dctx.ResourceGroupName = c.GetSessionVars().StmtCtx.ResourceGroupName
	return dctx
}

func TestFormatSQL(t *testing.T) {
	val := executor.FormatSQL("aaaa")
	require.Equal(t, "aaaa", val.String())
	variable.QueryLogMaxLen.Store(0)
	val = executor.FormatSQL("aaaaaaaaaaaaaaaaaaaa")
	require.Equal(t, "aaaaaaaaaaaaaaaaaaaa", val.String())
	variable.QueryLogMaxLen.Store(5)
	val = executor.FormatSQL("aaaaaaaaaaaaaaaaaaaa")
	require.Equal(t, "aaaaa(len:20)", val.String())
}

func TestContextCancelWhenReadFromCopIterator(t *testing.T) {
	store := testkit.CreateMockStore(t)
	tk := testkit.NewTestKit(t, store)
	tk.MustExec("use test")
	tk.MustExec("create table t(a int)")
	tk.MustExec("insert into t values(1)")

	syncCh := make(chan struct{})
	require.NoError(t, failpoint.EnableCall("github.com/pingcap/tidb/pkg/store/copr/CtxCancelBeforeReceive",
		func(ctx context.Context) {
			if ctx.Value("TestContextCancel") == "test" {
				syncCh <- struct{}{}
				<-syncCh
			}
		},
	))
	ctx := context.WithValue(context.Background(), "TestContextCancel", "test")
	ctx, cancelFunc := context.WithCancel(ctx)
	defer cancelFunc()
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		ctx = util.WithInternalSourceType(ctx, "scheduler")
		rs, err := tk.Session().ExecuteInternal(ctx, "select * from test.t")
		require.NoError(t, err)
		_, err2 := session.ResultSetToStringSlice(ctx, tk.Session(), rs)
		require.ErrorIs(t, err2, context.Canceled)
	}()
	<-syncCh
	cancelFunc()
	syncCh <- struct{}{}
	wg.Wait()
}

func TestFinishExecuteStmtReportsTiDBRUV2WithoutSyncingRUDetails(t *testing.T) {
	original := config.GetGlobalConfig()
	originalGenerateBinaryPlan := variable.GenerateBinaryPlan.Load()
	t.Cleanup(func() {
		if original != nil {
			config.StoreGlobalConfig(original)
		}
		variable.GenerateBinaryPlan.Store(originalGenerateBinaryPlan)
	})
	variable.GenerateBinaryPlan.Store(false)

	cfg := config.NewConfig()
	cfg.RUV2 = config.DefaultRUV2Config()
	cfg.Instance.EnableSlowLog.Store(false)
	cfg.Instance.RecordPlanInSlowLog = 0
	config.StoreGlobalConfig(cfg)

	reporter := &mockRUV2ConsumptionReporter{}
	ctx := &mockRUV2ReportingContext{
		Context:  mock.NewContext(),
		reporter: reporter,
	}
	sessVars := ctx.GetSessionVars()
	sessVars.StartTime = time.Now()
	sessVars.StmtCtx.StmtType = "Select"
	sessVars.StmtCtx.OriginalSQL = "select 1"
	sessVars.StmtCtx.ResetSQLDigest(sessVars.StmtCtx.OriginalSQL)
	sessVars.StmtCtx.ResourceGroupName = "rg1"

	goCtx := context.WithValue(context.Background(), execdetails.StmtExecDetailKey, &execdetails.StmtExecDetails{})
	goCtx = context.WithValue(goCtx, util.ExecDetailsKey, &util.ExecDetails{})
	goCtx = context.WithValue(goCtx, util.RUDetailsCtxKey, util.NewRUDetails())
	goCtx = context.WithValue(goCtx, execdetails.RUV2MetricsCtxKey, execdetails.NewRUV2Metrics())
	sessVars.RUV2Metrics = execdetails.RUV2MetricsFromContext(goCtx)
	require.NotNil(t, sessVars.RUV2Metrics)
	sessVars.RUV2Metrics.AddResultChunkCells(100)
	sessVars.RUV2Metrics.AddPlanCnt(2)
	sessVars.RUV2Metrics.AddSessionParserTotal(3)

	ruDetails := goCtx.Value(util.RUDetailsCtxKey).(*util.RUDetails)
	ruDetails.AddTiKVRUV2(23456)
	ruDetails.UpdateTiFlash(&rmpb.Consumption{RRU: 345, WRU: 67})
	commitDetails := &util.CommitDetails{
		WriteKeys: 3,
		WriteSize: 66,
	}
	sessVars.StmtCtx.SyncExecDetails.MergeExecDetails(nil, commitDetails)
	expectedMetrics := sessVars.RUV2Metrics.Clone()
	execdetails.UpdateRUV2MetricsFromCommitDetails(expectedMetrics, commitDetails)
	expected := expectedMetrics.CalculateRUValues(sessVars.RUV2Weights())

	execStmt := &executor.ExecStmt{
		Ctx:      ctx,
		GoCtx:    goCtx,
		StmtNode: &ast.SelectStmt{},
	}
	execStmt.FinishExecuteStmt(0, nil, false)

	require.Equal(t, float64(23456), ruDetails.TiKVRUV2())
	require.Equal(t, int64(3), sessVars.RUV2Metrics.WriteKeys())
	require.Equal(t, int64(66), sessVars.RUV2Metrics.WriteSize())
	require.Equal(t, "rg1", reporter.group)
	require.Equal(t, float64(23456), reporter.tikvRUV2)
	require.Equal(t, expected, reporter.tidbRUV2)
	require.Equal(t, float64(412), reporter.tiflashRU)

	t.Run("stmt summary ignores optimistic autocommit retry count", func(t *testing.T) {
		store := testkit.CreateMockStore(t)
		tk := testkit.NewTestKit(t, store)
		// Toggle stmt summary off and back on to clear any in-memory rows left by earlier tests.
		tk.MustExec("set global tidb_enable_stmt_summary = 0")
		tk.MustExec("set global tidb_enable_stmt_summary = 1")

		tk = testkit.NewTestKit(t, store)
		require.NoError(t, tk.Session().Auth(&auth.UserIdentity{Username: "root", Hostname: "%"}, nil, nil, nil))
		tk.MustExec("use test")
		tk.MustExec("set @@session.tidb_txn_mode = 'optimistic'")
		tk.MustExec("create table stmt_summary_retry (id int primary key, v int)")
		tk.MustExec("insert into stmt_summary_retry values (1, 1)")

		require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/session/mockCommitError8942", `1*return(true)->return(false)`))
		defer func() {
			require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/session/mockCommitError8942"))
		}()

		updateSQL := "update stmt_summary_retry set v = v + 1 where id = 1"
		tk.MustExec(updateSQL)
		tk.MustQuery(
			"select sum_exec_retry, sum_exec_retry_time from information_schema.statements_summary where digest_text like ?",
			"update `stmt_summary_retry`%",
		).Check(testkit.Rows("0 0"))
	})

	t.Run("bypass ru skips final reporting", func(t *testing.T) {
		reporter := &mockRUV2ConsumptionReporter{}
		ctx := &mockRUV2ReportingContext{
			Context:  mock.NewContext(),
			reporter: reporter,
		}
		sessVars := ctx.GetSessionVars()
		sessVars.StartTime = time.Now()
		sessVars.StmtCtx.StmtType = "Select"
		sessVars.StmtCtx.OriginalSQL = "select 1"
		sessVars.StmtCtx.ResetSQLDigest(sessVars.StmtCtx.OriginalSQL)
		sessVars.StmtCtx.ResourceGroupName = "rg1"

		goCtx := execdetails.ContextWithInitializedExecDetails(context.Background())
		sessVars.RUV2Metrics = execdetails.RUV2MetricsFromContext(goCtx)
		require.NotNil(t, sessVars.RUV2Metrics)
		sessVars.RUV2Metrics.SetBypass(true)
		sessVars.RUV2Metrics.AddResultChunkCells(100)

		ruDetails := goCtx.Value(util.RUDetailsCtxKey).(*util.RUDetails)
		ruDetails.AddTiKVRUV2(12345)
		ruDetails.UpdateTiFlash(&rmpb.Consumption{RRU: 10, WRU: 20})

		execStmt := &executor.ExecStmt{
			Ctx:      ctx,
			GoCtx:    goCtx,
			StmtNode: &ast.SelectStmt{},
		}
		execStmt.FinishExecuteStmt(0, nil, false)

		require.Empty(t, reporter.group)
		require.Zero(t, reporter.tikvRUV2)
		require.Zero(t, reporter.tidbRUV2)
		require.Zero(t, reporter.tiflashRU)
	})
}

func TestInsertRowsColMultiplyRUV2SQLPath(t *testing.T) {
	store := testkit.CreateMockStore(t)
	tk := testkit.NewTestKit(t, store)
	tk.MustExec("use test")
	tk.MustExec("create table t(a int primary key, b int, c int)")
	tk.MustExec("create table src(a int primary key, b int, c int)")
	tk.MustExec("insert into src values (10, 11, 12), (20, 21, 22)")

	runInsert := func(sql string) int64 {
		ctx := execdetails.ContextWithInitializedExecDetails(context.Background())
		tk.MustExecWithContext(ctx, sql)
		metrics := execdetails.RUV2MetricsFromContext(ctx)
		require.NotNil(t, metrics)
		return metrics.ExecutorL5InsertRows()
	}

	require.Equal(t, int64(6), runInsert("insert into t values (1, 2, 3), (2, 3, 4)"))
	require.Equal(t, int64(4), runInsert("insert into t(a, c) values (3, 5), (4, 6)"))
	require.Equal(t, int64(4), runInsert("insert into t(a, b) select a, b from src"))

	oldEnableBatchDML := variable.EnableBatchDML.Load()
	variable.EnableBatchDML.Store(true)
	defer variable.EnableBatchDML.Store(oldEnableBatchDML)

	tk.MustExec("set @@session.tidb_batch_insert=1")
	tk.MustExec("set @@session.tidb_dml_batch_size=2")
	tk.MustExec("create table batch_t(a int primary key, b int, c int)")
	tk.MustExec("insert into batch_t values (100, 100, 100)")

	ctx := execdetails.ContextWithInitializedExecDetails(context.Background())
	_, err := tk.ExecWithContext(ctx, "insert into batch_t values (1, 2, 3), (2, 3, 4), (100, 5, 6), (3, 4, 5)")
	require.Error(t, err)
	metrics := execdetails.RUV2MetricsFromContext(ctx)
	require.NotNil(t, metrics)
	require.Equal(t, int64(12), metrics.ExecutorL5InsertRows())
	tk.MustQuery("select a, b, c from batch_t order by a").Check(testkit.Rows(
		"1 2 3",
		"2 3 4",
		"100 100 100",
	))
}

func TestDMLRowsColMultiplyRUV2SQLPath(t *testing.T) {
	store := testkit.CreateMockStore(t)
	tk := testkit.NewTestKit(t, store)
	tk.MustExec("use test")
	tk.MustExec("create table t(a int primary key, b int, c int)")

	runDML := func(sql string) int64 {
		ctx := execdetails.ContextWithInitializedExecDetails(context.Background())
		tk.MustExecWithContext(ctx, sql)
		metrics := execdetails.RUV2MetricsFromContext(ctx)
		require.NotNil(t, metrics)
		return metrics.ExecutorL5InsertRows()
	}

	require.Equal(t, int64(6), runDML("replace into t values (1, 2, 3), (2, 3, 4)"))
	require.Equal(t, int64(6), runDML("update t set b = b + 10 where a in (1, 2)"))
	require.Equal(t, int64(3), runDML("delete from t where a = 1"))

	tk.MustExec("create table multi_del_l(a int primary key)")
	tk.MustExec("create table multi_del_r(a int primary key)")
	tk.MustExec("insert into multi_del_l values (1), (2)")
	tk.MustExec("insert into multi_del_r values (1), (2)")
	require.Equal(t, int64(4), runDML("delete multi_del_l, multi_del_r from multi_del_l join multi_del_r on multi_del_l.a = multi_del_r.a"))

	tk.MustExec("create table outer_l(a int primary key, b int)")
	tk.MustExec("create table outer_r(a int primary key, b int)")
	tk.MustExec("insert into outer_l values (1, 10), (2, 20)")
	tk.MustExec("insert into outer_r values (1, 100)")
	require.Equal(t, int64(2), runDML("update outer_l left join outer_r on outer_l.a = outer_r.a set outer_r.b = outer_r.b + 1"))

	tk.MustExec("create table dup_t(a int primary key, b int)")
	tk.MustExec("create table dup_s(a int, b int)")
	tk.MustExec("insert into dup_t values (1, 10)")
	tk.MustExec("insert into dup_s values (1, 100), (1, 200)")
	require.Equal(t, int64(2), runDML("update dup_t join dup_s on dup_t.a = dup_s.a set dup_t.b = dup_t.b + 1"))
}