compile

package
v0.0.0-debug-20260702 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 2, 2026 License: Apache-2.0 Imports: 162 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DistributedThreshold   uint64 = 10 * mpool.MB
	SingleLineSizeEstimate uint64 = 300 * mpool.B

	NoAccountId = -1
)

Note: Now the cost going from stat is actually the number of rows, so we can only estimate a number for the size of each row. The current insertion of around 200,000 rows triggers cn to write s3 directly

View Source 
const (
	Merge magicType = iota
	Normal
	Remote
	CreateDatabase
	CreateTable
	CreatePitr
	CreateCDC
	CreateView
	CreateIndex
	DropDatabase
	DropTable
	DropPitr
	DropCDC
	DropIndex
	TruncateTable
	AlterView
	AlterTable
	RenameTable
	MergeInsert
	MergeDelete
	CreateSequence
	DropSequence
	AlterSequence
	Replace
	TableClone
)

type of scope

View Source 
const (
	INDEX_TYPE_PRIMARY  = "PRIMARY"
	INDEX_TYPE_UNIQUE   = "UNIQUE"
	INDEX_TYPE_MULTIPLE = "MULTIPLE"
	INDEX_TYPE_FULLTEXT = "FULLTEXT"
	INDEX_TYPE_SPATIAL  = "SPATIAL"
)
View Source 
const (
	INDEX_VISIBLE_YES = 1
	INDEX_VISIBLE_NO  = 0
)
View Source 
const (
	INDEX_HIDDEN_YES = 1
	INDEX_HIDDEN_NO  = 0
)
View Source 
const (
	NULL_VALUE   = "null"
	EMPTY_STRING = ""
)
View Source 
const (
	ALLOCID_INDEX_KEY = "index_key"
)
View Source 
const MaxRpcTime = time.Hour * 24

MaxRpcTime is a default timeout time to rpc context if user never set this deadline. this is just a number I casually wrote, the purpose of doing this is that any message sent through rpc need a clear deadline.

View Source 
const StreamMaxInterval = 8192

Variables

View Source 
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
}
View Source 
var GetConstraintDef = func(ctx context.Context, rel engine.Relation) (*engine.ConstraintDef, error) {
	defs, err := rel.TableDefs(ctx)
	if err != nil {
		return nil, err
	}

	return GetConstraintDefFromTableDefs(defs), nil
}

Functions

func AddChildTblIdToParentTable added in v1.1.2 

func AddChildTblIdToParentTable(ctx context.Context, fkRelation engine.Relation, tblId uint64) error

func AddFkeyToRelation added in v1.1.2 

func AddFkeyToRelation(ctx context.Context, fkRelation engine.Relation, fkey *plan.ForeignKeyDef) error

func ApplyRuntimeFilters added in v1.0.0 

func ApplyRuntimeFilters(
	ctx context.Context,
	proc *process.Process,
	tableDef *plan.TableDef,
	relData engine.RelData,
	exprs []*plan.Expr,
	runtimeFilters []message.RuntimeFilterMessage,
) (engine.RelData, error)

func AttachInternalExecutorPrivilegeCheck 

func AttachInternalExecutorPrivilegeCheck(ctx context.Context) context.Context

AttachInternalExecutorPrivilegeCheck forces internal SQL to run through the normal privilege validation path instead of bypassing auth as trusted SQL.

func AttachInternalExecutorSession 

func AttachInternalExecutorSession(ctx context.Context, ses process.Session) context.Context

AttachInternalExecutorSession attaches the original frontend session to internal SQL so temp-table aliases and other session-scoped metadata resolve the same way they do for the user's statement.

func CDCParseGranularityTuple 

func CDCParseGranularityTuple(
	ctx context.Context,
	level string,
	pattern string,
	dup map[string]struct{},
) (pt *cdc.PatternTuple, err error)

func CDCParsePitrGranularity 

func CDCParsePitrGranularity(
	ctx context.Context,
	level string,
	tables string,
) (pts *cdc.PatternTuples, err error)

func CDCParseTableInfo 

func CDCParseTableInfo(
	ctx context.Context,
	input string,
	level string,
) (db string, table string, err error)

func CDCStrToTime 

func CDCStrToTime(tsStr string, tz *time.Location) (ts time.Time, err error)

func CheckSysMoCatalogPitrResult 

func CheckSysMoCatalogPitrResult(ctx context.Context, vecs []*vector.Vector, newLength uint64, newUnit string) (needInsert, needUpdate bool, err error)

CheckSysMoCatalogPitrResult parses the sys_mo_catalog_pitr query result and determines whether to insert or update. Arguments: 

ctx: context for error reporting
vecs: the vectors from the query result (should have at least 2 columns)
newLength: the new PITR length to compare
newUnit: the new PITR unit to compare

Returns: 

needInsert: true if sys_mo_catalog_pitr does not exist
needUpdate: true if it exists and needs update
oldLength, oldUnit: the old values if exist (for debug)
err: error if any

func CnServerMessageHandler added in v0.6.0 

func CnServerMessageHandler(
	ctx context.Context,
	serverAddress string,
	message morpc.Message,
	cs morpc.ClientSession,
	storageEngine engine.Engine, fileService fileservice.FileService, lockService lockservice.LockService,
	queryClient qclient.QueryClient,
	HaKeeper logservice.CNHAKeeperClient, udfService udf.Service, txnClient client.TxnClient,
	autoIncreaseCM *defines.AutoIncrCacheManager,
	messageAcquirer func() morpc.Message) (err error)

CnServerMessageHandler receive and deal the message from cn-client.

The message should always *pipeline.Message here. there are 2 types of pipeline message now.

  1. notify message : a message to tell the dispatch pipeline where its remote receiver are. and we use this connection's write-back method to send the data. or a message to stop the running pipeline.

  2. scope message : a message contains the encoded pipeline. we decoded it and run it locally.

func ConvertOperatorToPhyOperator 

func ConvertOperatorToPhyOperator(op vm.Operator, rmp map[*process.WaitRegister]int) *models.PhyOperator

ConvertOperatorToPhyOperator converts an Operator tree to a PhyOperator tree. All PhyOperator nodes and child-pointer slices are allocated from a single arena to minimize heap allocations and reduce GC pressure.

func ConvertScopeToPhyScope 

func ConvertScopeToPhyScope(scope *Scope, receiverMap map[*process.WaitRegister]int) models.PhyScope

func ConvertSourceToPhySource 

func ConvertSourceToPhySource(source *Source) *models.PhySource

func CreateAllIndexCdcTasks 

func CreateAllIndexCdcTasks(c *Compile, indexes []*plan.IndexDef, dbname string, tablename string, tableid uint64, startFromNow bool, tableDef *plan.TableDef) error

NOTE: CreateAllIndexCdcTasks will create CDC task according to existing tableDef

func CreateAllIndexUpdateTasks 

func CreateAllIndexUpdateTasks(c *Compile, indexes []*plan.IndexDef, dbname string, tablename string, tableid uint64) (err error)

idxcron function

func CreateCdcTask 

func CreateCdcTask(c *Compile, spec *iscp.JobSpec, job *iscp.JobID, startFromNow bool) (bool, error)

start here

func CreateIndexCdcTask 

func CreateIndexCdcTask(c *Compile, dbname string, tablename string, tableid uint64, indexname string, sinker_type int8, startFromNow bool, sql string, tableDef *plan.TableDef) error

NOTE: CreateIndexCdcTask will create CDC task without any checking. Original TableDef may be empty

func DebugShowScopes added in v0.6.0 

func DebugShowScopes(ss []*Scope, level DebugLevel) string

DebugShowScopes generates and returns a string representation of debugging information for a set of scopes.

func DecodeMergeGroup added in v1.1.0 

func DecodeMergeGroup(merge *group.MergeGroup, pipe *pipeline.Group)

func DeleteCdcTask 

func DeleteCdcTask(c *Compile, job *iscp.JobID) (bool, error)

func DropAllIndexCdcTasks 

func DropAllIndexCdcTasks(c *Compile, tabledef *plan.TableDef, dbname string, tablename string) error

drop all cdc tasks according to tableDef

func DropAllIndexUpdateTasks 

func DropAllIndexUpdateTasks(c *Compile, tabledef *plan.TableDef, dbname string, tablename string) (err error)

drop all cdc tasks according to tableDef

func DropIndexCdcTask 

func DropIndexCdcTask(c *Compile, tableDef *plan.TableDef, dbname string, tablename string, indexname string) error

func EncodeMergeGroup added in v1.1.0 

func EncodeMergeGroup(merge *group.MergeGroup, pipe *pipeline.Group)

func ExecuteAndGetRowsAffected 

func ExecuteAndGetRowsAffected(
	ctx context.Context,
	tx taskservice.SqlExecutor,
	query string,
	args ...interface{},
) (int64, error)

func GetConstraintDefFromTableDefs added in v1.1.2 

func GetConstraintDefFromTableDefs(defs []engine.TableDef) *engine.ConstraintDef

func GetExternParallelSize 

func GetExternParallelSize(totalSize int64, cpuNum int) int

func MakeNewCreateConstraint added in v1.1.2 

func MakeNewCreateConstraint(oldCt *engine.ConstraintDef, c engine.Constraint) (*engine.ConstraintDef, error)

func MarkQueryDone 

func MarkQueryDone(c *Compile, txn txnClient.TxnOperator)

func MarkQueryRunning 

func MarkQueryRunning(c *Compile, txn txnClient.TxnOperator)

func NewSQLExecutor added in v0.8.0 

func NewSQLExecutor(
	addr string,
	eng engine.Engine,
	mp *mpool.MPool,
	txnClient client.TxnClient,
	fs fileservice.FileService,
	qc qclient.QueryClient,
	hakeeper logservice.CNHAKeeperClient,
	us udf.Service,
	taskService taskservice.TaskService,
) executor.SQLExecutor

NewSQLExecutor returns a internal used sql service. It can execute sql in current CN.

func RegisterJob 

func RegisterJob(ctx context.Context, cnUUID string, txn client.TxnOperator, spec *iscp.JobSpec, job *iscp.JobID, startFromNow bool) (bool, error)

CDC APIs

func ReleaseScopes added in v1.1.0 

func ReleaseScopes(ss []*Scope)
func ShowPipelineLink(node vm.Operator, mp map[*process.WaitRegister]int, buffer *bytes.Buffer)

func ShowPipelineTree 

func ShowPipelineTree(
	node vm.Operator,
	prefix string,
	isRoot bool,
	isTail bool,
	mp map[*process.WaitRegister]int,
	level DebugLevel,
	buffer *bytes.Buffer,
)

func StrictSqlMode 

func StrictSqlMode(proc *process.Process) (error, bool)

func UnregisterJob 

func UnregisterJob(ctx context.Context, cnUUID string, txn client.TxnOperator, job *iscp.JobID) (bool, error)

func UpdatePreparePhyOperator 

func UpdatePreparePhyOperator(op vm.Operator, phyOp *models.PhyOperator) bool

func UpdatePreparePhyScope 

func UpdatePreparePhyScope(scope *Scope, phyScope models.PhyScope) bool

func UpdateScopeTxnOffset 

func UpdateScopeTxnOffset(scope *Scope, txnOffset int)

Types

type AnalyzeModule 

type AnalyzeModule struct {
	// contains filtered or unexported fields
}

func (*AnalyzeModule) AppendRemotePhyPlan 

func (anal *AnalyzeModule) AppendRemotePhyPlan(remotePhyPlan models.PhyPlan)

func (*AnalyzeModule) GetExplainPhyBuffer 

func (anal *AnalyzeModule) GetExplainPhyBuffer() *bytes.Buffer

func (*AnalyzeModule) GetPhyPlan 

func (anal *AnalyzeModule) GetPhyPlan() *models.PhyPlan

func (*AnalyzeModule) Reset 

func (anal *AnalyzeModule) Reset(isPrepare bool, isTpQuery bool)

Reset When Compile reused, reset AnalyzeModule to prevent resource accumulation

func (*AnalyzeModule) TypeName 

func (anal *AnalyzeModule) TypeName() string

type CDCCreateTaskOptions 

type CDCCreateTaskOptions struct {
	TaskName     string
	TaskId       string
	UserInfo     *CDCUserInfo
	Exclude      string
	StartTs      string
	EndTs        string
	MaxSqlLength int64
	PitrTables   string // json encoded pitr tables: cdc2.PatternTuples
	SrcUri       string // json encoded source uri: cdc2.UriInfo
	SrcUriInfo   cdc.UriInfo
	SinkUri      string // json encoded sink uri: cdc2.UriInfo
	SinkUriInfo  cdc.UriInfo
	ExtraOpts    string // json encoded extra opts: map[string]any
	SinkType     string
	NoFull       bool
	ConfigFile   string

	// control options
	UseConsole bool
}

func (*CDCCreateTaskOptions) BuildTaskDetails 

func (opts *CDCCreateTaskOptions) BuildTaskDetails() (details *task.Details, err error)

func (*CDCCreateTaskOptions) BuildTaskMetadata 

func (opts *CDCCreateTaskOptions) BuildTaskMetadata() task.TaskMetadata

func (*CDCCreateTaskOptions) ToInsertTaskSQL 

func (opts *CDCCreateTaskOptions) ToInsertTaskSQL(
	ctx context.Context,
	tx taskservice.SqlExecutor,
) (sql string, err error)

func (*CDCCreateTaskOptions) ValidateAndFill 

func (opts *CDCCreateTaskOptions) ValidateAndFill(
	ctx context.Context,
	c *Compile,
	planCDC *plan.CreateCDC,
) (err error)

type CDCUserInfo 

type CDCUserInfo struct {
	UserName    string
	AccountId   uint32
	AccountName string
}

type Col 

type Col struct {
	Typ  types.T
	Name string
}

Col is the information of attribute

type Compile added in v0.5.0 

type Compile struct {

	// TxnOffset read starting offset position within the transaction during the execute current statement
	TxnOffset int

	MessageBoard *message.MessageBoard
	// contains filtered or unexported fields
}

Compile contains all the information needed for compilation.

func NewCompile added in v1.1.0 

func NewCompile(
	addr, db, sql, tenant, uid string,
	e engine.Engine,
	proc *process.Process,
	stmt tree.Statement,
	isInternal bool,
	cnLabel map[string]string,
	startAt time.Time,
) *Compile

NewCompile is used to new an object of compile

func (*Compile) AnalyzeExecPlan 

func (c *Compile) AnalyzeExecPlan(runC *Compile, queryResult *util2.RunResult, stats *statistic.StatsInfo, isExplainPhy bool, option *ExplainOption)

func (*Compile) Compile added in v0.5.0 

func (c *Compile) Compile(
	execTopContext context.Context,
	queryPlan *plan.Plan,
	resultWriteBack func(batch *batch.Batch, crs *perfcounter.CounterSet) error) (err error)

Compile generates the node level execution pipeline from the query plan, and the final pipeline will be stored in the attribute `scope` of a Compile object.

func (*Compile) FreeOperator 

func (c *Compile) FreeOperator()

func (*Compile) GenPhyPlan 

func (c *Compile) GenPhyPlan(runC *Compile)

func (*Compile) GetAnalyzeModule 

func (c *Compile) GetAnalyzeModule() *AnalyzeModule

func (*Compile) GetMessageCenter 

func (c *Compile) GetMessageCenter() *message.MessageCenter

func (*Compile) GetPlan 

func (c *Compile) GetPlan() *plan.Plan

GetPlan returns the current plan of the Compile. This is useful for getting the latest plan after a retry.

func (*Compile) InitPipelineContextToExecuteQuery 

func (c *Compile) InitPipelineContextToExecuteQuery()

InitPipelineContextToExecuteQuery initializes the context for each pipeline tree.

the entire process must follow these rules: 1. the query context can control the context of all pipelines. 2. if there's a data transfer between two pipelines, the lifecycle of the sender's context ends with the receiver's termination.

func (*Compile) InitPipelineContextToRetryQuery 

func (c *Compile) InitPipelineContextToRetryQuery()

InitPipelineContextToRetryQuery initializes the context for each pipeline tree. the only place diff to InitPipelineContextToExecuteQuery is this function build query context from the last query.

func (*Compile) IsSingleScope 

func (c *Compile) IsSingleScope(ss []*Scope) bool

func (*Compile) IsTpQuery 

func (c *Compile) IsTpQuery() bool

func (*Compile) Release added in v1.1.0 

func (c *Compile) Release()

func (*Compile) Reset 

func (c *Compile) Reset(proc *process.Process, startAt time.Time, fill func(*batch.Batch, *perfcounter.CounterSet) error, sql string)

func (*Compile) Run added in v0.5.0 

func (c *Compile) Run(_ uint64) (queryResult *util2.RunResult, err error)

Run executes the pipeline and returns the result.

func (*Compile) SetBuildPlanFunc added in v1.0.0 

func (c *Compile) SetBuildPlanFunc(buildPlanFunc func(ctx context.Context) (*plan2.Plan, error))

func (*Compile) SetIsPrepare 

func (c *Compile) SetIsPrepare(isPrepare bool)

func (*Compile) SetOriginSQL added in v1.2.0 

func (c *Compile) SetOriginSQL(sql string)

func (Compile) TypeName added in v1.2.0 

func (c Compile) TypeName() string

func (*Compile) UpdatePreparePhyPlan 

func (c *Compile) UpdatePreparePhyPlan(runC *Compile) bool

true means update success, if return false, GenPhyPlan

type DebugLevel 

type DebugLevel int
const (
	NormalLevel DebugLevel = iota
	VerboseLevel
	AnalyzeLevel
	OldLevel
)

type ExplainOption 

type ExplainOption struct {
	Verbose bool
	Analyze bool
}

type LockMeta 

type LockMeta struct {
	// contains filtered or unexported fields
}

func NewLockMeta 

func NewLockMeta() *LockMeta

type MultiTableIndex added in v1.2.0 

type MultiTableIndex struct {
	IndexAlgo string
	IndexDefs map[string]*plan.IndexDef
}

type ParallelScopeInfo 

type ParallelScopeInfo struct {
	NodeIdxTimeConsumeMajor map[int]int64
	NodeIdxTimeConsumeMinor map[int]int64
}

func NewParallelScopeInfo 

func NewParallelScopeInfo() *ParallelScopeInfo

type RemoteReceivRegInfo added in v0.7.0 

type RemoteReceivRegInfo struct {
	Idx      int
	Uuid     uuid.UUID
	FromAddr string
}

type RuntimeFilterEvaluator added in v1.0.0 

type RuntimeFilterEvaluator interface {
	Evaluate(objectio.ZoneMap) bool
}

type RuntimeInFilter added in v1.0.0 

type RuntimeInFilter struct {
	InList *vector.Vector
}

func (*RuntimeInFilter) Evaluate added in v1.0.0 

func (f *RuntimeInFilter) Evaluate(zm objectio.ZoneMap) bool

type RuntimeZonemapFilter added in v1.0.0 

type RuntimeZonemapFilter struct {
	Zm objectio.ZoneMap
}

func (*RuntimeZonemapFilter) Evaluate added in v1.0.0 

func (f *RuntimeZonemapFilter) Evaluate(zm objectio.ZoneMap) bool

type Scope 

type Scope struct {
	// Magic specifies the type of Scope.
	// 0 -  execution unit for reading data.
	// 1 -  execution unit for processing intermediate results.
	// 2 -  execution unit that requires remote call.
	Magic magicType

	// IsEnd means the pipeline is end
	IsEnd bool

	// IsRemote means the pipeline is remote
	IsRemote bool

	// IsLoad means the pipeline is load
	IsLoad bool

	// IsTbFunc means the leaf op of pipeline is tablefunction, tablefunction is src op
	IsTbFunc bool

	HasPartialResults bool
	// StarCountOnly: when true, aggOptimize took the single-starcount fast path.
	// buildReaders should return EmptyReaders and no data should flow.
	StarCountOnly bool
	// StarCountMergeGroup: set when StarCountOnly is true; resetForReuse clears its PartialResults.
	StarCountMergeGroup *group.MergeGroup

	Plan *plan.Plan
	// DataSource stores information about data source.
	DataSource *Source
	// PreScopes contains children of this scope will inherit and execute.
	PreScopes []*Scope
	// NodeInfo contains the information about the remote node.
	NodeInfo engine.Node
	// TxnOffset represents the transaction's write offset, specifying the starting position for reading data.
	TxnOffset int
	// Instructions contains command list of this scope.
	// Instructions vm.Instructions
	RootOp vm.Operator
	// Proc contains the execution context.
	Proc *process.Process

	ScopeAnalyzer *ScopeAnalyzer

	RemoteReceivRegInfos []RemoteReceivRegInfo
}

Scope is the output of the compile process. Each sql will be compiled to one or more execution unit scopes.

func (*Scope) AlterSequence added in v1.0.0 

func (s *Scope) AlterSequence(c *Compile) error

func (*Scope) AlterTable added in v0.8.0 

func (s *Scope) AlterTable(c *Compile) (err error)

func (*Scope) AlterTableCopy added in v1.0.0 

func (s *Scope) AlterTableCopy(c *Compile) error

func (*Scope) AlterTableInplace added in v1.0.0 

func (s *Scope) AlterTableInplace(c *Compile) error

func (*Scope) AlterView added in v0.7.0 

func (s *Scope) AlterView(c *Compile) error

Drop the old view, and create the new view.

func (*Scope) CreateCDC 

func (s *Scope) CreateCDC(c *Compile) error

func (*Scope) CreateDatabase 

func (s *Scope) CreateDatabase(c *Compile) error

func (*Scope) CreateIndex 

func (s *Scope) CreateIndex(c *Compile) error

func (*Scope) CreatePitr 

func (s *Scope) CreatePitr(c *Compile) error

func (*Scope) CreateSequence added in v0.8.0 

func (s *Scope) CreateSequence(c *Compile) error

func (*Scope) CreateTable 

func (s *Scope) CreateTable(c *Compile) error

func (*Scope) CreateView added in v1.2.1 

func (s *Scope) CreateView(c *Compile) error

func (*Scope) DropCDC 

func (s *Scope) DropCDC(c *Compile) error

func (*Scope) DropDatabase 

func (s *Scope) DropDatabase(c *Compile) error

func (*Scope) DropIndex 

func (s *Scope) DropIndex(c *Compile) error

func (*Scope) DropPitr 

func (s *Scope) DropPitr(c *Compile) error

func (*Scope) DropSequence added in v0.8.0 

func (s *Scope) DropSequence(c *Compile) error

func (*Scope) DropTable 

func (s *Scope) DropTable(c *Compile) error

func (*Scope) FreeOperator 

func (s *Scope) FreeOperator(c *Compile)

func (*Scope) InitAllDataSource added in v1.2.0 

func (s *Scope) InitAllDataSource(c *Compile) error

func (*Scope) IsTableClone 

func (s *Scope) IsTableClone() bool

IsTableClone reports whether this scope executes a `create table … clone` — the statement snapshot/restore replays to rebuild a table. Restore-aware behavior in the compile/plugin layer keys off this: the experimental-flag gate below, and pluginCompileCtx.IsTableClone exposed to index plugins.

func (*Scope) MergeRun 

func (s *Scope) MergeRun(c *Compile) error

MergeRun case 1 : 

specific run for Tp query without merge operator.

case 2 : 

normal merge run.
1. start n goroutines from pool to run the pre-scope.
2. send notify message to remote node for its data producer.
3. run itself.
4. listen to all running pipelines, once any error occurs, stop the NormalMergeRun asap.

func (*Scope) ParallelRun 

func (s *Scope) ParallelRun(c *Compile) (err error)

ParallelRun run a pipeline in parallel.

func (*Scope) RemoteRun 

func (s *Scope) RemoteRun(c *Compile) error

RemoteRun send the scope to a remote node for execution.

func (*Scope) RenameTable 

func (s *Scope) RenameTable(c *Compile) (err error)

func (*Scope) Reset 

func (s *Scope) Reset(c *Compile) error

func (*Scope) RestoreTable 

func (s *Scope) RestoreTable(c *Compile, clonePlan *plan.CloneTable) error

RestoreTable is the clone/restore-path twin of cloneUnaffectedIndexes (pkg/sql/compile/alter.go). CreateTable (already run by TableClone) seeds the index hidden tables and registers their CDC; the block-level clone in table_clone APPENDS onto those tables. So, in place of a bare s.Run(c):

  1. drop the index CDC tasks before cloning data;
  2. for each hidden table the plugin lists in DeleteBeforeClone (IVF-FLAT's metadata/centroids/entries — seeded non-empty by CreateTable), empty the seed with `DELETE … WHERE TRUE` (a content delete that keeps the table and its id — NOT truncate, which re-creates the table);
  3. s.Run(c): clone the main table + index hidden tables (append onto empty);
  4. re-register each index's CDC startFromNow=true with a PLUGIN-PROVIDED InitSQL. For a vector index that InitSQL is `ALTER … REINDEX … FORCE_SYNC`, so the CDC's first iteration runs the reindex in its own post-commit txn — rebuilding the model from the committed cloned rows and re-arming the CDC at the post-clone watermark. Running it as InitSQL (not inline in this clone txn) is what avoids the SnapshotTS replay that double-counts the cloned rows.

func (*Scope) Run 

func (s *Scope) Run(c *Compile) (err error)

Run read data from storage engine and run the instructions of scope. Note: The prepare time for executing the `scope`.`Run()` method is very short, and no statistics are done

func (*Scope) SetOperatorInfoRecursively added in v1.2.0 

func (s *Scope) SetOperatorInfoRecursively(cb func() int32)

func (*Scope) TableClone 

func (s *Scope) TableClone(c *Compile) error

func (*Scope) TruncateTable added in v0.6.0 

func (s *Scope) TruncateTable(c *Compile) error

func (Scope) TypeName added in v1.2.0 

func (s Scope) TypeName() string

type ScopeAnalyzer 

type ScopeAnalyzer struct {
	TimeConsumed int64 // Stores the total time consumed between Start and Stop in nanoseconds
	// contains filtered or unexported fields
}

func NewScopeAnalyzer 

func NewScopeAnalyzer() *ScopeAnalyzer

func (*ScopeAnalyzer) Reset 

func (sa *ScopeAnalyzer) Reset()

Reset clears the analyzer's state, allowing it to start again. Both isStarted and isStoped flags are reset.

func (*ScopeAnalyzer) Start 

func (sa *ScopeAnalyzer) Start()

Start begins the time tracking. It will not start if it has already started or if it has been stopped.

func (*ScopeAnalyzer) Stop 

func (sa *ScopeAnalyzer) Stop()

Stop halts the time tracking and calculates the duration. It won't perform any actions if it has not started or if it has already been stopped.

type Source 

type Source struct {
	PushdownId      uint64
	PushdownAddr    string
	SchemaName      string
	RelationName    string
	Attributes      []string
	R               engine.Reader
	Rel             engine.Relation
	FilterExpr      *plan.Expr   // todo: change this to []*plan.Expr,  is FilterList + RuntimeFilter
	FilterList      []*plan.Expr //from node.FilterList, use for reader
	BlockFilterList []*plan.Expr //from node.BlockFilterList, use for range

	TableDef  *plan.TableDef
	Timestamp timestamp.Timestamp
	AccountId *plan.PubInfo

	RuntimeFilterSpecs []*plan.RuntimeFilterSpec
	OrderBy            []*plan.OrderBySpec // for ordered scan

	IndexReaderParam *plan.IndexReaderParam

	RecvMsgList           []plan.MsgHeader
	MembershipFilterBytes []byte
	// contains filtered or unexported fields
}

Source contains information of a relation which will be used in execution.

type TxnOperator added in v0.6.0 

type TxnOperator = client.TxnOperator

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.