frontend

var (
	// the sqls creating many tables for the tenant.
	// Wrap them in a transaction
	MoCatalogMoUserDDL = `` /* 573-byte string literal not displayed */

	MoCatalogMoAccountDDL = `` /* 380-byte string literal not displayed */

	MoCatalogMoRoleDDL = `` /* 220-byte string literal not displayed */

	MoCatalogMoUserGrantDDL = `` /* 182-byte string literal not displayed */

	MoCatalogMoRoleGrantDDL = `` /* 262-byte string literal not displayed */

	MoCatalogMoRolePrivsDDL = `` /* 407-byte string literal not displayed */

	MoCatalogMoUserDefinedFunctionDDL = `` /* 580-byte string literal not displayed */

	MoCatalogMoMysqlCompatibilityModeDDL = `` /* 323-byte string literal not displayed */

	MoCatalogMoSnapshotsDDL = fmt.Sprintf(`CREATE TABLE %s.%s (
			snapshot_id uuid unique key,
			sname varchar(64) primary key,
			ts bigint,
			level enum('cluster','account','database','table'),
	        account_name varchar(300),
			database_name varchar(5000),
			table_name  varchar(5000),
			obj_id bigint unsigned,
    		kind varchar(32) not null default 'user'
			)`, catalog.MO_CATALOG, catalog.MO_SNAPSHOTS)

	MoCatalogMoPitrDDL = fmt.Sprintf(`CREATE TABLE %s.%s (
			pitr_id uuid unique key,
			pitr_name varchar(5000),
			create_account bigint unsigned,
			create_time bigint not null,
			modified_time bigint not null,
			level varchar(10),
			account_id bigint unsigned,
			account_name varchar(300),
			database_name varchar(5000),
			table_name varchar(5000),
			obj_id bigint unsigned,
			pitr_length tinyint unsigned,
			pitr_unit varchar(10),
			pitr_status tinyint unsigned default 1 comment '1: active, 0: inactive',
			pitr_status_changed_time bigint not null,
    		kind varchar(32) not null default 'user',
			primary key(pitr_name, create_account)
			)`, catalog.MO_CATALOG, catalog.MO_PITR)

	MoCatalogMoPubsDDL = `` /* 449-byte string literal not displayed */

	MoCatalogMoSubsDDL = `` /* 562-byte string literal not displayed */

	MoCatalogMoStoredProcedureDDL = `` /* 551-byte string literal not displayed */

	MoCatalogMoStagesDDL = `` /* 265-byte string literal not displayed */

	MoCatalogMoCdcTaskDDL = `` /* 1061-byte string literal not displayed */

	MoCatalogMoCdcWatermarkDDL = `` /* 279-byte string literal not displayed */

	MoCatalogMoISCPLogDDL = `` /* 428-byte string literal not displayed */

	MoCatalogMoCcprLogDDL = `` /* 681-byte string literal not displayed */

	MoCatalogMoCcprTablesDDL = fmt.Sprintf(`CREATE TABLE %s.%s (
				tableid BIGINT UNSIGNED PRIMARY KEY,
				taskid UUID NOT NULL,
				dbname VARCHAR(256) NOT NULL,
				tablename VARCHAR(256) NOT NULL,
				account_id INT UNSIGNED NOT NULL
			)`, catalog.MO_CATALOG, catalog.MO_CCPR_TABLES)

	MoCatalogMoCcprDbsDDL = fmt.Sprintf(`CREATE TABLE %s.%s (
				dbid BIGINT UNSIGNED PRIMARY KEY,
				taskid UUID NOT NULL,
				dbname VARCHAR(256) NOT NULL,
				account_id INT UNSIGNED NOT NULL
			)`, catalog.MO_CATALOG, catalog.MO_CCPR_DBS)

	MoCatalogMoIndexUpdateDDL = `` /* 591-byte string literal not displayed */

	MoCatalogMoSessionsDDL       = `` /* 277-byte string literal not displayed */
	MoCatalogMoConfigurationsDDL = `` /* 165-byte string literal not displayed */
	MoCatalogMoLocksDDL          = `` /* 156-byte string literal not displayed */
	MoCatalogMoVariablesDDL      = `` /* 191-byte string literal not displayed */
	MoCatalogMoTransactionsDDL   = `` /* 244-byte string literal not displayed */
	MoCatalogMoCacheDDL          = `CREATE VIEW mo_catalog.mo_cache AS SELECT node_type, node_id, type, used, free, hit_ratio FROM mo_cache() AS mo_cache_tmp`

	MoCatalogMoDataKeyDDL = `` /* 328-byte string literal not displayed */

	MoCatalogMoTableStatsDDL = fmt.Sprintf(`create table mo_catalog.%s (
    			account_id bigint signed,
    			database_id bigint signed,
    			table_id bigint signed,
    			database_name varchar(255),
    			table_name varchar(255),
    			table_stats json,
    			update_time datetime(6) not null,
    			takes bigint unsigned,
    			primary key(account_id, database_id, table_id)
			)`, catalog.MO_TABLE_STATS)

	MoCatalogMoAccountLockDDL = fmt.Sprintf(`create table mo_catalog.%s(
    			account_name varchar(300) primary key
				)`, catalog.MO_ACCOUNT_LOCK)

	MoCatalogMergeSettingsDDL = fmt.Sprintf(`create table mo_catalog.%s (
		account_id int unsigned not null,
		tid bigint unsigned not null,
		version int unsigned not null,
		settings json not null,
		extra_info text,
		primary key(account_id, tid)
	)`, catalog.MO_MERGE_SETTINGS)

	MoCatalogMergeSettingsInitData = fmt.Sprintf(`insert into mo_catalog.%s values (
		0, 0, %d, '%s', '')`,
		catalog.MO_MERGE_SETTINGS,
		merge.MergeSettingsVersion_Curr,
		merge.DefaultMergeSettings.String())

	MoCatalogBranchMetadataDDL = fmt.Sprintf(`create table mo_catalog.%s(
    	table_id bigint unsigned comment 'id of the table this branch points to',
   	 	clone_ts bigint signed not null comment 'branch creation timestamp in nanoseconds',
    	p_table_id bigint unsigned not null comment 'id of the parent table this branch is based on',
    	creator bigint unsigned not null comment 'account id of the creator',
    	level varchar not null,
    	table_deleted bool not null default false,
    	index(p_table_id),
    	index(creator),
    	primary key(table_id)
	)`, catalog.MO_BRANCH_METADATA)

	MoCatalogFeatureLimitDDL = fmt.Sprintf(`create table mo_catalog.%s(
    	account_id bigint unsigned not null comment 'this limit applies on this account',
    	feature_code varchar(50) NOT NULL comment 'snapshot/branch/...',
		scope varchar(50) NOT NULL DEFAULT '' comment 'feature limit applies on this scope',
    	quota bigint NOT NULL DEFAULT 100 comment '0: disabled this feature; -1 unlimited; >0: max allowed value',
    	created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
		updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    	primary key(account_id, feature_code, scope)
	)`, catalog.MO_FEATURE_LIMIT)

	MoCatalogFeatureRegistryDDL = fmt.Sprintf(`create table mo_catalog.%s(
    	feature_code varchar(50) NOT NULL comment 'snapshot/branch/...',
    	description varchar(1024) NOT NULL DEFAULT '',
        scope_spec JSON NOT NULL comment 'allowed scope values',
		enabled boolean NOT NULL DEFAULT TRUE,
    	created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
		updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    	primary key(feature_code)
	)`, catalog.MO_FEATURE_REGISTRY)

	MoCatalogFeatureRegistryInitData = fmt.Sprintf(`insert into mo_catalog.%s(feature_code, scope_spec) values 
		('SNAPSHOT', '{"allowed_scope":["account","database","table"]}'),
		('BRANCH', '{"allowed_scope":[]}')
		on duplicate key update scope_spec = values(scope_spec);`, catalog.MO_FEATURE_REGISTRY)

	MoCatalogMoRoleRuleDDL = fmt.Sprintf("create table %s.%s ("+
		"role_id int signed,"+
		"rule_name varchar(200),"+
		"`rule` text,"+
		"primary key(role_id, rule_name)"+
		")", catalog.MO_CATALOG, catalog.MO_ROLE_RULE)
)

var (
	MoCatalogMoAutoIncrTableDDL = fmt.Sprintf(`create table %s.%s (
			table_id   bigint unsigned, 
			col_name   varchar(770), 
			col_index  int,
			offset     bigint unsigned, 
			step       bigint unsigned,  
			primary key(table_id, col_name)
		)`, catalog.MO_CATALOG, catalog.MOAutoIncrTable)

	MoCatalogMoIndexesDDL = fmt.Sprintf(`create table %s.%s (
			id                  bigint unsigned not null,
			table_id            bigint unsigned not null,
			database_id         bigint unsigned not null,
			name                varchar(64) not null,
			type                varchar(11) not null,
			algo                varchar(11),
			algo_table_type     varchar(11),
			algo_params         varchar(2048),
			is_visible          tinyint not null,
			hidden              tinyint not null,
			comment             varchar(2048) not null,
			column_name         varchar(256) not null,
			ordinal_position    int unsigned  not null,
			options             text,
			index_table_name    varchar(5000),
			primary key(id, column_name)
		)`, catalog.MO_CATALOG, catalog.MO_INDEXES)

	MoCatalogMoForeignKeysDDL = fmt.Sprintf(`create table %s.%s (
			constraint_name varchar(5000) not null,
			constraint_id BIGINT UNSIGNED not null default 0,
			db_name varchar(5000) not null,
			db_id BIGINT UNSIGNED not null default 0,
			table_name varchar(5000) not null,
			table_id BIGINT UNSIGNED not null default 0,
			column_name varchar(256) not null,
			column_id BIGINT UNSIGNED not null default 0,
			refer_db_name varchar(5000) not null,
			refer_db_id BIGINT UNSIGNED not null default 0,
			refer_table_name varchar(5000) not null,
			refer_table_id BIGINT UNSIGNED not null default 0,
			refer_column_name varchar(256) not null,
			refer_column_id BIGINT UNSIGNED not null default 0,
			on_delete varchar(128) not null,
			on_update varchar(128) not null,
	
			primary key(
				constraint_name,
				constraint_id,
				db_name,
				db_id,
				table_name,
				table_id,
				column_name,
				column_id,
				refer_db_name,
				refer_db_id,
				refer_table_name,
				refer_table_id,
				refer_column_name,
				refer_column_id)
		)`, catalog.MO_CATALOG, catalog.MOForeignKeys)

	MoCatalogMoTablePartitionsDDL = fmt.Sprintf(`CREATE TABLE %s.%s (
			table_id bigint unsigned NOT NULL,
			database_id bigint unsigned not null,
			number smallint unsigned NOT NULL,
			name varchar(64) NOT NULL,
			partition_type varchar(50) NOT NULL,
			partition_expression varchar(2048) NULL,
			description_utf8 text,
			comment varchar(2048) NOT NULL,
			options text,
			partition_table_name varchar(1024) NOT NULL,
			PRIMARY KEY table_id (table_id, name)
		)`, catalog.MO_CATALOG, catalog.MO_TABLE_PARTITIONS)
)

var (
	MoCatalogMoVersionDDL = fmt.Sprintf(`create table %s.%s (
			version             varchar(50) not null,
		    version_offset      int unsigned default 0,
			state               int,
			create_at           timestamp not null,
			update_at           timestamp not null,
			primary key(version, version_offset)
		)`, catalog.MO_CATALOG, catalog.MOVersionTable)

	MoCatalogMoUpgradeDDL = fmt.Sprintf(`create table %s.%s (
			id                   bigint unsigned not null primary key auto_increment,
			from_version         varchar(50) not null,
			to_version           varchar(50) not null,
			final_version        varchar(50) not null,
            final_version_offset int unsigned default 0,
			state                int,
			upgrade_cluster      int,
			upgrade_tenant       int,
			upgrade_order        int,
			total_tenant         int,
			ready_tenant         int,
			create_at            timestamp not null,
			update_at            timestamp not null
		)`, catalog.MO_CATALOG, catalog.MOUpgradeTable)

	MoCatalogMoUpgradeTenantDDL = fmt.Sprintf(`create table %s.%s (
			id                  bigint unsigned not null primary key auto_increment,
			upgrade_id		    bigint unsigned not null,
			target_version      varchar(50) not null,
			from_account_id     int not null,
			to_account_id       int not null,
			ready               int,
			create_at           timestamp not null,
			update_at           timestamp not null
		)`, catalog.MO_CATALOG, catalog.MOUpgradeTenantTable)
)

var (
	MoTaskSysAsyncTaskDDL = fmt.Sprintf(`create table %s.sys_async_task (
			task_id                     bigint primary key auto_increment,
			task_metadata_id            varchar(50) not null,
			task_metadata_executor      int,
			task_metadata_context       blob,
			task_metadata_option        varchar(1000),
			task_parent_id              varchar(50),
			task_status                 int,
			task_runner                 varchar(50),
			task_epoch                  int,
			last_heartbeat              bigint,
			result_code                 int null,
			error_msg                   varchar(1000) null,
			create_at                   bigint,
			end_at                      bigint)`,
		catalog.MOTaskDB)

	MoTaskSysCronTaskDDL = fmt.Sprintf(`create table %s.sys_cron_task (
			cron_task_id				bigint primary key auto_increment,
    		task_metadata_id            varchar(50) unique not null,
			task_metadata_executor      int,
			task_metadata_context       blob,
			task_metadata_option 		varchar(1000),
			cron_expr					varchar(100) not null,
			next_time					bigint,
			trigger_times				int,
			create_at					bigint,
			update_at					bigint)`,
		catalog.MOTaskDB)

	MoTaskSysDaemonTaskDDL = fmt.Sprintf(`create table %s.sys_daemon_task (
				task_id                     bigint primary key auto_increment,
				task_metadata_id            varchar(50),
			task_metadata_executor      int,
			task_metadata_context       blob,
			task_metadata_option        varchar(1000),
			account_id                  int unsigned not null,
			account                     varchar(128) not null,
			task_type                   varchar(64) not null,
			task_runner                 varchar(64),
			task_status                 int not null,
			last_heartbeat              timestamp,
			create_at                   timestamp not null,
			update_at                   timestamp not null,
				end_at                      timestamp,
				last_run                    timestamp,
				details                     blob)`,
		catalog.MOTaskDB)

	MoTaskSQLTaskDDL = fmt.Sprintf(`create table %s.%s (
				task_id                     bigint primary key auto_increment,
				task_name                   varchar(128) not null,
				account_id                  int unsigned not null,
				database_name               varchar(128) not null default '',
				cron_expr                   varchar(128) not null default '',
				`+"`timezone`"+`                  varchar(64) not null default 'UTC',
				sql_body                    text not null,
				gate_condition              text not null default '',
				retry_limit                 int not null default 0,
				timeout_seconds             int not null default 0,
				enabled                     tinyint not null default 1,
				next_fire_time              bigint not null default 0,
				trigger_count               bigint not null default 0,
				creator                     varchar(128) not null default '',
				creator_user_id             int unsigned not null default 0,
				creator_role_id             int unsigned not null default 0,
				created_at                  timestamp not null default current_timestamp,
				updated_at                  timestamp not null default current_timestamp,
				unique key uk_task_name_account (task_name, account_id))`,
		catalog.MOTaskDB, catalog.MOSQLTask)

	MoTaskSQLTaskRunDDL = fmt.Sprintf(`create table %s.%s (
				run_id                      bigint primary key auto_increment,
				task_id                     bigint not null,
				task_name                   varchar(128) not null,
				account_id                  int unsigned not null,
				scheduled_at                timestamp null,
				started_at                  timestamp null,
				finished_at                 timestamp null,
				duration_seconds            double not null default 0,
				status                      varchar(20) not null default 'RUNNING',
				trigger_type                varchar(20) not null default 'SCHEDULED',
				attempt_number              int not null default 1,
				rows_affected               bigint not null default 0,
				error_code                  int not null default 0,
				error_message               text,
				gate_result                 tinyint not null default 1,
				runner_cn                   varchar(128) not null default '',
				index idx_task_id (task_id),
				index idx_status (status),
				index idx_started_at (started_at))`,
		catalog.MOTaskDB, catalog.MOSQLTaskRun)
)

var AllocNewBlock = func(c *Conn, allocLength int) error {
	return c.AllocNewBlock(allocLength)
}

var AppendCountOfBytesLenEnc = func(mp *MysqlProtocolImpl, value []byte) error {
	return mp.appendCountOfBytesLenEnc(value)
}

var AppendPart = func(c *Conn, elems []byte) error {
	return c.AppendPart(elems)
}

var AppendStringLenEnc = func(mp *MysqlProtocolImpl, value string) error {
	return mp.appendStringLenEnc(value)
}

var BeginPacket = func(c *Conn) error {
	return c.BeginPacket()
}

var CDCCheckPitrGranularity = func(
	ctx context.Context,
	bh BackgroundExec,
	accName string,
	pts *cdc.PatternTuples,
	minLength ...int64,
) error {
	var minPitrLen int64 = 2
	if len(minLength) > 1 {
		return moerr.NewInternalErrorf(ctx, "only one length parameter allowed")
	}
	if len(minLength) > 0 {
		minPitrLen = max(minLength[0]+1, minPitrLen)
	}

	getPitrConfig := func(level, dbName, tblName string) (*PitrConfig, error) {
		config := NewPitrConfig(level)
		length, unit, ok, err := getPitrLengthAndUnit(ctx, bh, level, accName, dbName, tblName)
		if err != nil {
			return nil, err
		}
		config.Length = length
		config.Unit = unit
		config.Exists = ok
		return config, nil
	}

	if config, err := getPitrConfig(cdc.CDCPitrGranularity_Cluster, "", ""); err != nil {
		return err
	} else if config.IsValid(minPitrLen) {
		return nil
	}

	for _, pt := range pts.Pts {
		dbName := pt.Source.Database
		tblName := pt.Source.Table

		level := cdc.CDCPitrGranularity_Table
		if dbName == cdc.CDCPitrGranularity_All && tblName == cdc.CDCPitrGranularity_All {
			level = cdc.CDCPitrGranularity_Account
		} else if tblName == cdc.CDCPitrGranularity_All {
			level = cdc.CDCPitrGranularity_DB
		}

		if config, err := getPitrConfig(cdc.CDCPitrGranularity_Account, dbName, tblName); err != nil {
			return err
		} else if config.IsValid(minPitrLen) {
			continue
		}

		if level == cdc.CDCPitrGranularity_DB || level == cdc.CDCPitrGranularity_Table {
			if config, err := getPitrConfig(cdc.CDCPitrGranularity_DB, dbName, tblName); err != nil {
				return err
			} else if config.IsValid(minPitrLen) {
				continue
			}
		}

		if level == cdc.CDCPitrGranularity_Table {
			if config, err := getPitrConfig(cdc.CDCPitrGranularity_Table, dbName, tblName); err != nil {
				return err
			} else if config.IsValid(minPitrLen) {
				continue
			}
		}

		return moerr.NewInternalErrorf(ctx,
			"no valid PITR configuration found for pattern: %s, minimum required length: %d hours",
			pt.OriginString, minPitrLen)
	}
	return nil
}

var CDCExectorError_QueryDaemonTaskTimeout = moerr.NewInternalErrorNoCtx("query daemon task timeout")

var CDCExeutorAllocator *mpool.MPool

var CDCShowOutputColumns = [CDCShowOutputColumnsCount]Column{
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameTaskID_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameTaskID_Idx].columnType,
		},
	},
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameTaskName_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameTaskName_Idx].columnType,
		},
	},
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameSourceURI_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameSourceURI_Idx].columnType,
		},
	},
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameSinkURI_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameSinkURI_Idx].columnType,
		},
	},
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameState_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameState_Idx].columnType,
		},
	},
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameErrMsg_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameErrMsg_Idx].columnType,
		},
	},
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameCheckpoint_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameCheckpoint_Idx].columnType,
		},
	},
	&MysqlColumn{
		ColumnImpl: ColumnImpl{
			name:       CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameTimestamp_Idx].name,
			columnType: CDCShowOutputColumnsSpec[CDCShowOutputColumnsNameTimestamp_Idx].columnType,
		},
	},
}

var CDCShowOutputColumnsSpec = map[int]struct {
	name       string
	columnType defines.MysqlType
}{
	CDCShowOutputColumnsNameTaskID_Idx:     {/* contains filtered or unexported fields */},
	CDCShowOutputColumnsNameTaskName_Idx:   {/* contains filtered or unexported fields */},
	CDCShowOutputColumnsNameSourceURI_Idx:  {/* contains filtered or unexported fields */},
	CDCShowOutputColumnsNameSinkURI_Idx:    {/* contains filtered or unexported fields */},
	CDCShowOutputColumnsNameState_Idx:      {/* contains filtered or unexported fields */},
	CDCShowOutputColumnsNameErrMsg_Idx:     {/* contains filtered or unexported fields */},
	CDCShowOutputColumnsNameCheckpoint_Idx: {/* contains filtered or unexported fields */},
	CDCShowOutputColumnsNameTimestamp_Idx:  {/* contains filtered or unexported fields */},
}

var Close = func(ep *ExportConfig) error {
	ep.FileCnt++
	err := ep.AsyncWriter.Close()
	if err != nil {
		return err
	}
	err = ep.AsyncGroup.Wait()
	if err != nil {
		return err
	}
	err = ep.AsyncReader.Close()
	if err != nil {
		return err
	}
	ep.AsyncReader = nil
	ep.AsyncWriter = nil
	ep.AsyncGroup = nil
	return err
}

var Collations []*Collation = []*Collation{
	{"utf8_general_ci", "utf8", 33, "", "Yes", 1, "PAD SPACE"},
	{"binary", "binary", 63, "YES", "Yes", 1, "NO PAD"},
	{"utf8_unicode_ci", "utf8", 192, "", "Yes", 1, "PAD SPACE"},
	{"utf8_bin", "utf8", 83, "YES", "Yes", 1, "NO PAD"},
	{"utf8mb4_general_ci", "utf8mb4", 45, "", "Yes", 1, "PAD SPACE"},
	{"utf8mb4_unicode_ci", "utf8mb4", 224, "", "Yes", 1, "PAD SPACE"},
	{"utf8mb4_bin", "utf8mb4", 46, "YES", "Yes", 1, "NO PAD"},
	{"utf8mb4_0900_bin", "utf8mb4", 309, "", "Yes", 1, "NO PAD"},
	{"utf8mb4_0900_ai_ci", "utf8mb4", 255, "", "Yes", 0, "NO PAD"},
	{"utf8mb4_de_pb_0900_ai_ci", "utf8mb4", 256, "", "Yes", 0, "NO PAD"},
	{"utf8mb4_is_0900_ai_ci", "utf8mb4", 257, "", "Yes", 0, "NO PAD"},
	{"utf8mb4_lv_0900_ai_ci", "utf8mb4", 258, "", "Yes", 0, "NO PAD"},
}

var ComputeDdlBatchWithSnapshotFunc = computeDdlBatchWithSnapshotImpl

var ConnIDAllocKey = "____server_conn_id"

var DefaultCapability = CLIENT_LONG_PASSWORD |
	CLIENT_FOUND_ROWS |
	CLIENT_LONG_FLAG |
	CLIENT_CONNECT_WITH_DB |
	CLIENT_LOCAL_FILES |
	CLIENT_PROTOCOL_41 |
	CLIENT_INTERACTIVE |
	CLIENT_TRANSACTIONS |
	CLIENT_SECURE_CONNECTION |
	CLIENT_MULTI_STATEMENTS |
	CLIENT_MULTI_RESULTS |
	CLIENT_PLUGIN_AUTH |
	CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA |
	CLIENT_DEPRECATE_EOF |
	CLIENT_CONNECT_ATTRS

var DefaultClientConnStatus = SERVER_STATUS_AUTOCOMMIT

var ExeSqlInBgSes = func(reqCtx context.Context, bh BackgroundExec, sql string) ([]ExecResult, error) {
	return executeSQLInBackgroundSession(reqCtx, bh, sql)
}

var FinishedPacket = func(c *Conn) error {
	return c.FinishedPacket()
}

var ForeachQueriedRow = func(
	ctx context.Context,
	tx taskservice.SqlExecutor,
	query string,
	onEachRow func(context.Context, *sql.Rows) (bool, error),
) (cnt int64, err error) {
	var (
		ok   bool
		rows *sql.Rows
	)
	if rows, err = tx.QueryContext(ctx, query); err != nil {
		return
	}
	if rows.Err() != nil {
		err = rows.Err()
		return
	}
	defer func() {
		_ = rows.Close()
	}()

	for rows.Next() {
		if ok, err = onEachRow(ctx, rows); err != nil {
			return
		}
		if ok {
			cnt++
		}
	}
	return
}

var GSysVarsMgr = &GlobalSysVarsMgr{
	accountsGlobalSysVarsMap: make(map[uint32]*SystemVariables),
}

var GetBool = func(slices *ColumnSlices, rowIdx uint64, colIdx uint64) (bool, error) {
	return slices.GetBool(rowIdx, colIdx)
}

var GetBytesBased = func(slices *ColumnSlices, r uint64, i uint64) ([]byte, error) {
	return slices.GetBytesBased(r, i)
}

var GetComputationWrapper = func(execCtx *ExecCtx, db string, user string, eng engine.Engine, proc *process.Process, ses *Session) ([]ComputationWrapper, error) {
	var cws []ComputationWrapper = nil
	if preparePlan := execCtx.input.getPreparePlan(); preparePlan != nil {
		tcw := InitTxnComputationWrapper(ses, execCtx.input.stmt, proc)
		tcw.plan = preparePlan.GetDcl().GetPrepare().Plan
		tcw.binaryPrepare = execCtx.input.isBinaryProtExecute
		tcw.prepareName = execCtx.input.stmtName
		cws = append(cws, tcw)
		return cws, nil
	} else if cached := ses.getCachedPlan(execCtx.input.getHash()); cached != nil {
		for i, stmt := range cached.stmts {
			tcw := InitTxnComputationWrapper(ses, stmt, proc)
			tcw.plan = cached.plans[i]
			cws = append(cws, tcw)
		}

		return cws, nil
	}

	var stmts []tree.Statement = nil
	var cmdFieldStmt *InternalCmdFieldList
	var cmdGetSnapshotTsStmt *InternalCmdGetSnapshotTs
	var cmdGetDatabasesStmt *InternalCmdGetDatabases
	var cmdGetMoIndexesStmt *InternalCmdGetMoIndexes
	var cmdGetDdlStmt *InternalCmdGetDdl
	var cmdGetObjectStmt *InternalCmdGetObject
	var cmdObjectListStmt *InternalCmdObjectList
	var err error

	if execCtx.input.getStmt() != nil {
		stmts = append(stmts, execCtx.input.getStmt())
	} else if isCmdFieldListSql(execCtx.input.getSql()) {
		cmdFieldStmt, err = parseCmdFieldList(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdFieldStmt)
	} else if isCmdGetSnapshotTsSql(execCtx.input.getSql()) {
		cmdGetSnapshotTsStmt, err = parseCmdGetSnapshotTs(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetSnapshotTsStmt)
	} else if isCmdGetDatabasesSql(execCtx.input.getSql()) {
		cmdGetDatabasesStmt, err = parseCmdGetDatabases(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetDatabasesStmt)
	} else if isCmdGetMoIndexesSql(execCtx.input.getSql()) {
		cmdGetMoIndexesStmt, err = parseCmdGetMoIndexes(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetMoIndexesStmt)
	} else if isCmdGetDdlSql(execCtx.input.getSql()) {
		cmdGetDdlStmt, err = parseCmdGetDdl(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetDdlStmt)
	} else if isCmdGetObjectSql(execCtx.input.getSql()) {
		cmdGetObjectStmt, err = parseCmdGetObject(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetObjectStmt)
	} else if isCmdObjectListSql(execCtx.input.getSql()) {
		cmdObjectListStmt, err = parseCmdObjectList(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdObjectListStmt)
	} else if isCmdCheckSnapshotFlushedSql(execCtx.input.getSql()) {
		cmdCheckSnapshotFlushedStmt, err := parseCmdCheckSnapshotFlushed(execCtx.reqCtx, execCtx.input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdCheckSnapshotFlushedStmt)
	} else {
		stmts, err = parseSql(execCtx, ses.GetMySQLParser())
		if err != nil {
			return nil, err
		}
		v, err := ses.GetSessionSysVar("enable_remap_hint")
		if err == nil {
			if on, convErr := valueIsBoolTrue(v); convErr == nil && on {
				err = parsers.AddRewriteHints(execCtx.reqCtx, stmts, execCtx.input.getSql())
				if err != nil {
					return nil, err
				}
			}
		}
	}

	for _, stmt := range stmts {
		cws = append(cws, InitTxnComputationWrapper(ses, stmt, proc))
	}
	return cws, nil
}

var GetComputationWrapperInBack = func(execCtx *ExecCtx, db string, input *UserInput, user string, eng engine.Engine, proc *process.Process, ses FeSession) ([]ComputationWrapper, error) {
	var cw []ComputationWrapper = nil

	var stmts []tree.Statement = nil
	var cmdFieldStmt *InternalCmdFieldList
	var cmdGetSnapshotTsStmt *InternalCmdGetSnapshotTs
	var cmdGetDatabasesStmt *InternalCmdGetDatabases
	var cmdGetMoIndexesStmt *InternalCmdGetMoIndexes
	var cmdGetDdlStmt *InternalCmdGetDdl
	var cmdGetObjectStmt *InternalCmdGetObject
	var cmdObjectListStmt *InternalCmdObjectList
	var err error

	if input.getStmt() != nil {
		stmts = append(stmts, input.getStmt())
	} else if isCmdFieldListSql(input.getSql()) {
		cmdFieldStmt, err = parseCmdFieldList(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdFieldStmt)
	} else if isCmdGetSnapshotTsSql(input.getSql()) {
		cmdGetSnapshotTsStmt, err = parseCmdGetSnapshotTs(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetSnapshotTsStmt)
	} else if isCmdGetDatabasesSql(input.getSql()) {
		cmdGetDatabasesStmt, err = parseCmdGetDatabases(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetDatabasesStmt)
	} else if isCmdGetMoIndexesSql(input.getSql()) {
		cmdGetMoIndexesStmt, err = parseCmdGetMoIndexes(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetMoIndexesStmt)
	} else if isCmdGetDdlSql(input.getSql()) {
		cmdGetDdlStmt, err = parseCmdGetDdl(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetDdlStmt)
	} else if isCmdGetObjectSql(input.getSql()) {
		cmdGetObjectStmt, err = parseCmdGetObject(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdGetObjectStmt)
	} else if isCmdObjectListSql(input.getSql()) {
		cmdObjectListStmt, err = parseCmdObjectList(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdObjectListStmt)
	} else if isCmdCheckSnapshotFlushedSql(input.getSql()) {
		cmdCheckSnapshotFlushedStmt, err := parseCmdCheckSnapshotFlushed(execCtx.reqCtx, input.getSql())
		if err != nil {
			return nil, err
		}
		stmts = append(stmts, cmdCheckSnapshotFlushedStmt)
	} else {
		stmts, err = parseSql(execCtx, ses.GetMySQLParser())
		if err != nil {
			return nil, err
		}
	}

	for _, stmt := range stmts {
		cw = append(cw, InitTxnComputationWrapper(ses, stmt, proc))
	}
	return cw, nil
}

var GetDate = func(slices *ColumnSlices, r uint64, i uint64) (types.Date, error) {
	return slices.GetDate(r, i)
}

var GetDatetime = func(slices *ColumnSlices, r uint64, i uint64) (string, error) {
	return slices.GetDatetime(r, i)
}

var GetDecimal = func(slices *ColumnSlices, r uint64, i uint64) (string, error) {
	return slices.GetDecimal(r, i)
}

var GetFloat32 = func(slices *ColumnSlices, r uint64, i uint64) (float32, error) {
	return slices.GetFloat32(r, i)
}

var GetFloat64 = func(slices *ColumnSlices, r uint64, i uint64) (float64, error) {
	return slices.GetFloat64(r, i)
}

var GetInt64 = func(slices *ColumnSlices, r uint64, i uint64) (int64, error) {
	return slices.GetInt64(r, i)
}

var GetObjectDataReader = func(ctx context.Context, ses *Session, objectName string, offset int64, size int64) ([]byte, error) {
	return readObjectFromFS(ctx, ses, objectName, offset, size)
}

var GetObjectFSProvider = func(ses *Session) (fileservice.FileService, error) {
	eng := getPu(ses.GetService()).StorageEngine
	var de *disttae.Engine
	var ok bool
	if de, ok = eng.(*disttae.Engine); !ok {
		var entireEngine *engine.EntireEngine
		if entireEngine, ok = eng.(*engine.EntireEngine); ok {
			de, ok = entireEngine.Engine.(*disttae.Engine)
		}
		if !ok {
			return nil, moerr.NewInternalErrorNoCtx("failed to get disttae engine")
		}
	}

	fs := de.FS()
	if fs == nil {
		return nil, moerr.NewInternalErrorNoCtx("fileservice is not available")
	}
	return fs, nil
}

var GetObjectPermissionChecker = func(ctx context.Context, ses *Session, pubAccountName, pubName string) (uint64, error) {
	if len(pubAccountName) == 0 || len(pubName) == 0 {
		return 0, moerr.NewInternalError(ctx, "publication account name and publication name are required for GET OBJECT")
	}
	bh := ses.GetShareTxnBackgroundExec(ctx, false)
	defer bh.Close()
	currentAccount := ses.GetTenantInfo().GetTenant()
	accountID, _, err := getAccountFromPublication(ctx, bh, pubAccountName, pubName, currentAccount)
	return accountID, err
}

var GetStringBased = func(slices *ColumnSlices, r uint64, i uint64) (string, error) {
	return slices.GetStringBased(r, i)
}

var GetTableErrMsg = func(
	ctx context.Context,
	accountId uint32,
	ieExecutor ie.InternalExecutor,
	taskId string,
	tbl *cdc.DbTableInfo) (
	hasError bool, err error,
) {
	ctx = defines.AttachAccountId(ctx, catalog.System_Account)
	sql := cdc.CDCSQLBuilder.GetTableErrMsgSQL(uint64(accountId), taskId, tbl.SourceDbName, tbl.SourceTblName)
	res := ieExecutor.Query(ctx, sql, ie.SessionOverrideOptions{})
	if res.Error() != nil {
		return false, res.Error()
	} else if res.RowCount() < 1 {
		return false, nil
	}

	errMsg, err := res.GetString(ctx, 0, 0)
	if err != nil {
		return false, err
	}
	if errMsg == "" {
		return false, nil
	}

	metadata := cdc.ParseErrorMetadata(errMsg)
	if metadata == nil {
		return false, nil
	}

	if cdc.ShouldRetry(metadata) {

		if metadata.IsRetryable {
			logutil.Info(
				"cdc.frontend.task.retryable_table_error",
				zap.String("db", tbl.SourceDbName),
				zap.String("table", tbl.SourceTblName),
				zap.Int("retry-count", metadata.RetryCount),
				zap.Int("max-retry", cdc.MaxRetryCount),
				zap.String("message", metadata.Message),
			)
		} else {

			age := time.Since(metadata.FirstSeen)
			logutil.Info(
				"cdc.frontend.task.expired_non_retryable_error",
				zap.String("db", tbl.SourceDbName),
				zap.String("table", tbl.SourceTblName),
				zap.Duration("age", age),
				zap.String("message", metadata.Message),
			)
		}
		return false, nil
	}

	if metadata.IsRetryable {

		logutil.Warn(
			"cdc.frontend.task.max_retry_exceeded",
			zap.String("db", tbl.SourceDbName),
			zap.String("table", tbl.SourceTblName),
			zap.Int("retry-count", metadata.RetryCount),
			zap.String("message", metadata.Message),
		)
	} else {

		age := time.Since(metadata.FirstSeen)
		logutil.Info(
			"cdc.frontend.task.permanent_table_error",
			zap.String("db", tbl.SourceDbName),
			zap.String("table", tbl.SourceTblName),
			zap.Duration("age", age),
			zap.String("message", metadata.Message),
		)
	}

	hasError = true
	return
}

var GetTime = func(slices *ColumnSlices, r uint64, i uint64) (string, error) {
	return slices.GetTime(r, i)
}

var GetTimestamp = func(slices *ColumnSlices, r uint64, i uint64, timeZone *time.Location) (string, error) {
	return slices.GetTimestamp(r, i, timeZone)
}

var GetUUID = func(slices *ColumnSlices, r uint64, i uint64) (string, error) {
	return slices.GetUUID(r, i)
}

var GetUint64 = func(slices *ColumnSlices, r uint64, i uint64) (uint64, error) {
	return slices.GetUint64(r, i)
}

var IsNull = func(slices *ColumnSlices, rowIdx int, colIdx int) bool {
	return slices.IsNull(rowIdx, colIdx)
}

var (
	MaxPrepareNumberInOneSession atomic.Uint32
)

var NewBackgroundExec = func(
	reqCtx context.Context,
	upstream FeSession,
	opts ...*BackgroundExecOption,
) BackgroundExec {
	// XXXSP
	var txnOp TxnOperator

	return upstream.InitBackExec(txnOp, "", backSesOutputCallback, opts...)
}

var NewShareTxnBackgroundExec = func(ctx context.Context, ses FeSession, rawBatch bool) BackgroundExec {
	return ses.GetShareTxnBackgroundExec(ctx, rawBatch)
}

var ObjectListPermissionChecker = func(ctx context.Context, ses *Session, pubAccountName, pubName string) (uint64, error) {
	if len(pubAccountName) == 0 || len(pubName) == 0 {
		return 0, moerr.NewInternalError(ctx, "publication account name and publication name are required for OBJECT LIST")
	}
	bh := ses.GetShareTxnBackgroundExec(ctx, false)
	defer bh.Close()
	currentAccount := ses.GetTenantInfo().GetTenant()
	accountID, _, err := getAccountFromPublication(ctx, bh, pubAccountName, pubName, currentAccount)
	return accountID, err
}

var PathExists = func(path string) (bool, bool, error) {
	fi, err := os.Stat(path)
	if err == nil {
		return true, !fi.IsDir(), nil
	}
	if os.IsNotExist(err) {
		return false, false, err
	}

	return false, false, err
}

var ProcessObjectListFunc = processObjectListImpl

var ReadNBytesIntoBuf = func(c *Conn, buf []byte, n int) error {
	return c.ReadNBytesIntoBuf(buf, n)
}

var ReadOnePayload = func(c *Conn, payloadLength int) ([]byte, error) {
	return c.ReadOnePayload(payloadLength)
}

var RecordParseErrorStatement = func(ctx context.Context, ses *Session, proc *process.Process, envBegin time.Time,
	envStmt []string, sqlTypes []string, err error) (context.Context, error) {
	retErr := moerr.NewParseError(ctx, err.Error())

	sqlType := ""
	if len(sqlTypes) > 0 {
		sqlType = sqlTypes[0]
	} else {
		sqlType = constant.ExternSql
	}

	if len(envStmt) > 0 {
		for i, sql := range envStmt {
			if i < len(sqlTypes) {
				sqlType = sqlTypes[i]
			}
			ctx, err = RecordStatement(ctx, ses, proc, nil, envBegin, sql, sqlType, true)
			if err != nil {
				return nil, err
			}
			ses.tStmt.EndStatement(ctx, retErr, 0, 0, 0)
		}
	} else {
		ctx, err = RecordStatement(ctx, ses, proc, nil, envBegin, "", sqlType, true)
		if err != nil {
			return nil, err
		}
		ses.tStmt.EndStatement(ctx, retErr, 0, 0, 0)
	}
	ses.SetTStmt(nil)

	tenant := ses.GetTenantInfo()
	if tenant == nil {
		tenant, _ = GetTenantInfo(ctx, "internal")
	}
	incStatementErrorsCounter(tenant.GetTenant(), tenant.GetTenantID(), nil)
	return ctx, nil
}

var RecordStatement = func(ctx context.Context, ses *Session, proc *process.Process, cw ComputationWrapper, envBegin time.Time, envStmt, sqlType string, useEnv bool) (context.Context, error) {
	// set StatementID
	var stmID uuid.UUID
	var statement tree.Statement = nil
	var text string
	if cw != nil {
		copy(stmID[:], cw.GetUUID())
		statement = cw.GetAst()

		ses.ast = statement
		binExec, prepareName := cw.BinaryExecute()
		execSql := makeExecuteSql(ctx, ses, statement, binExec, prepareName)
		if len(execSql) != 0 {
			bb := strings.Builder{}
			bb.WriteString(envStmt)
			bb.WriteString(" // ")
			bb.WriteString(execSql)
			text = commonutil.Abbreviate(bb.String(), int(getPu(ses.GetService()).SV.LengthOfQueryPrinted))
		} else {

			text = commonutil.Abbreviate(envStmt, int(getPu(ses.GetService()).SV.LengthOfQueryPrinted))
		}
	} else {
		u, _ := util.FastUuid()
		stmID = uuid.UUID(u)
		text = commonutil.Abbreviate(envStmt, int(getPu(ses.GetService()).SV.LengthOfQueryPrinted))
	}
	ses.SetStmtId(stmID)
	stmtTyp := getStatementType(statement).GetStatementType()
	queryTyp := getStatementType(statement).GetQueryType()
	ses.SetStmtType(stmtTyp)
	ses.SetQueryType(queryTyp)
	ses.SetSqlSourceType(sqlType)
	ses.SetSqlOfStmt(text)
	if proc != nil {

		proc.SetStmtProfile(&ses.stmtProfile)
	}
	ses.stmtProfile.SetDivByZeroRuntimeProfile(stmtTyp, queryTyp, isIgnoreStatement(statement))

	if sqlType != constant.InternalSql {
		ses.pushQueryId(types.Uuid(stmID).String())
	}

	if !motrace.GetTracerProvider().IsEnable() {
		return ctx, nil
	}
	if sqlType == constant.InternalSql && envStmt == "" {

		return ctx, nil
	}

	tenant := ses.GetTenantInfo()
	if tenant == nil {
		tenant, _ = GetTenantInfo(ctx, "internal")
	}
	stm := motrace.NewStatementInfo()
	// set TransactionID
	var txn TxnOperator
	var err error

	if handler := ses.GetTxnHandler(); handler.InActiveTxn() {
		txn = handler.GetTxn()
		if err != nil {
			return nil, err
		}
		stm.SetTxnID(txn.Txn().ID)
	}

	copy(stm.SessionID[:], ses.GetUUID())
	copy(stm.StatementID[:], stmID[:])
	requestAt := envBegin
	if !useEnv {
		requestAt = time.Now()
	}

	stm.ConnectionId = ses.GetConnectionID()
	stm.Account = tenant.GetTenant()
	stm.AccountID = tenant.GetTenantID()
	stm.RoleId = tenant.GetDefaultRoleID()

	stm.User = tenant.GetUser()
	stm.Host = ses.respr.GetStr(PEER)
	stm.Database = ses.respr.GetStr(DBNAME)
	stm.StatementFingerprint = ""
	stm.StatementTag = ""
	stm.SqlSourceType = sqlType
	stm.RequestAt = requestAt
	stm.StatementType = getStatementType(statement).GetStatementType()
	stm.QueryType = getStatementType(statement).GetQueryType()
	stm.ConnType = transferSessionConnType2StatisticConnType(ses.connType)
	if sqlType == constant.InternalSql && isCmdFieldListSql(envStmt) {

		stm.User = ""
	}
	if ses.disableAgg {
		stm.DisableAgg()
	}

	stm.RecordStatementSql(text, envStmt)
	stm.Report(ctx)
	ses.SetTStmt(stm)

	return ctx, nil
}

var RecordStatementTxnID = func(ctx context.Context, fses FeSession) error {
	var ses *Session
	var ok bool
	if ses, ok = fses.(*Session); !ok {
		return nil
	}
	var txn TxnOperator
	var err error
	if ses == nil {
		return nil
	}

	if stm := ses.tStmt; stm != nil && stm.IsZeroTxnID() {
		if handler := ses.GetTxnHandler(); handler.InActiveTxn() {

			txn = handler.GetTxn()
			if err != nil {
				return err
			}
			stm.SetTxnID(txn.Txn().ID)
			ses.SetTxnId(txn.Txn().ID)
		}

	}

	if upSes := ses.upstream; upSes != nil && upSes.tStmt != nil && upSes.tStmt.IsZeroTxnID() {

		if handler := ses.GetTxnHandler(); handler.InActiveTxn() {
			txn = handler.GetTxn()
			if err != nil {
				return err
			}

			if stmt := upSes.tStmt; stmt.NeedSkipTxn() {

				stmt.SetSkipTxn(false)
				stmt.SetSkipTxnId(txn.Txn().ID)
			} else if txnId := txn.Txn().ID; !stmt.SkipTxnId(txnId) {
				upSes.tStmt.SetTxnID(txnId)
			}
		}
	}
	return nil
}

var Write = func(ep *ExportConfig, output []byte) (int, error) {
	n, err := ep.AsyncWriter.Write(output)
	if err != nil {
		err2 := ep.AsyncWriter.CloseWithError(err)
		if err2 != nil {
			return 0, err2
		}
	}
	return n, err
}