testutil

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: 65 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func CreateEngines 

func CreateEngines(
	ctx context.Context,
	opts TestOptions,
	t *testing.T,
	funcOpts ...TestDisttaeEngineOptions,
) (
	disttaeEngine *TestDisttaeEngine,
	taeEngine *TestTxnStorage,
	rpcAgent *MockRPCAgent,
	mp *mpool.MPool,
)

func EmptyBatchFromSchema 

func EmptyBatchFromSchema(schema *catalog.Schema, sels ...int) *batch.Batch

func EndThisStatement 

func EndThisStatement(
	ctx context.Context,
	txn client.TxnOperator,
) (err error)

func EngineDefAddIndex 

func EngineDefAddIndex(defs []engine.TableDef, idxColName string) []engine.TableDef

func EngineTableDefBySchema 

func EngineTableDefBySchema(schema *catalog.Schema) ([]engine.TableDef, error)

func GetDefaultTNShard 

func GetDefaultTNShard() metadata.TNShard

func GetDefaultTestPath 

func GetDefaultTestPath(module string, t *testing.T) string

func GetRelationReader 

func GetRelationReader(
	ctx context.Context,
	e *TestDisttaeEngine,
	txn client.TxnOperator,
	relation engine.Relation,
	exprs []*plan.Expr,
	mp *mpool.MPool,
	t *testing.T,
) (reader engine.Reader, err error)

func GetTableTxnReader 

func GetTableTxnReader(
	ctx context.Context,
	e *TestDisttaeEngine,
	dbName, tableName string,
	exprs []*plan.Expr,
	mp *mpool.MPool,
	t *testing.T,
) (
	txn client.TxnOperator,
	relation engine.Relation,
	reader engine.Reader,
	err error,
)

func InitTxnHandle 

func InitTxnHandle(ctx context.Context, taeDir string, opts *options.Options) *rpc.Handle

func MakeDefaultTestPath 

func MakeDefaultTestPath(module string, t *testing.T) string

func MakeTxnHeartbeatMonkeyJob 

func MakeTxnHeartbeatMonkeyJob(
	e *TestTxnStorage,
	opInterval time.Duration,
) *tasks.CancelableJob

func MockShardService 

func MockShardService() *mockShardService

func NewDefaultTableReader 

func NewDefaultTableReader(
	ctx context.Context,
	rel engine.Relation,
	expr *plan.Expr,
	mp *mpool.MPool,
	ranges engine.RelData,
	snapshotTS timestamp.Timestamp,
	e *disttae.Engine,
	txnOffset int,
) (engine.Reader, error)

func PlanTableDefBySchema 

func PlanTableDefBySchema(schema *catalog.Schema, tableId uint64, databaseName string) plan.TableDef

func TxnRanges 

func TxnRanges(
	ctx context.Context,
	txn client.TxnOperator,
	relation engine.Relation,
	exprs []*plan.Expr,
) (engine.RelData, error)

func WriteToRelation 

func WriteToRelation(
	ctx context.Context,
	txn client.TxnOperator,
	relation engine.Relation,
	bat *batch.Batch,
	isDelete, toEndStatement bool,
) (err error)

Types

type EnginePack 

type EnginePack struct {
	D   *TestDisttaeEngine
	T   *TestTxnStorage
	R   *MockRPCAgent
	Mp  *mpool.MPool
	Ctx context.Context
	// contains filtered or unexported fields
}

func InitEnginePack 

func InitEnginePack(opts TestOptions, t *testing.T) *EnginePack

func (*EnginePack) Close 

func (p *EnginePack) Close()

func (*EnginePack) CreateDB 

func (p *EnginePack) CreateDB(txnop client.TxnOperator, dbname string) engine.Database

func (*EnginePack) CreateDBAndTable 

func (p *EnginePack) CreateDBAndTable(txnop client.TxnOperator, dbname string, schema *catalog.Schema) (engine.Database, engine.Relation)

func (*EnginePack) CreateDBAndTables 

func (p *EnginePack) CreateDBAndTables(txnop client.TxnOperator, dbname string, schema ...*catalog.Schema) (engine.Database, []engine.Relation)

func (*EnginePack) CreateTableInDB 

func (p *EnginePack) CreateTableInDB(txnop client.TxnOperator, dbname string, schema *catalog.Schema) engine.Relation

func (*EnginePack) DeleteTableInDB 

func (p *EnginePack) DeleteTableInDB(txnop client.TxnOperator, dbname, tblname string)

func (*EnginePack) StartCNTxn 

func (p *EnginePack) StartCNTxn(opts ...client.TxnOption) client.TxnOperator

type MockLogtailPRCServer 

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

func (*MockLogtailPRCServer) Close 

func (s *MockLogtailPRCServer) Close() error

func (*MockLogtailPRCServer) RegisterRequestHandler 

func (s *MockLogtailPRCServer) RegisterRequestHandler(
	onMessage func(ctx context.Context, request morpc.RPCMessage, sequence uint64, cs morpc.ClientSession) error)

func (*MockLogtailPRCServer) Start 

func (s *MockLogtailPRCServer) Start() error

type MockLogtailRPCClient 

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

func (*MockLogtailRPCClient) Close 

func (c *MockLogtailRPCClient) Close() error

func (*MockLogtailRPCClient) CloseBackend 

func (c *MockLogtailRPCClient) CloseBackend() error

func (*MockLogtailRPCClient) NewStream 

func (c *MockLogtailRPCClient) NewStream(ctx context.Context, backend string, lock bool) (morpc.Stream, error)

func (*MockLogtailRPCClient) Ping 

func (c *MockLogtailRPCClient) Ping(ctx context.Context, backend string) error

func (*MockLogtailRPCClient) Send 

func (c *MockLogtailRPCClient) Send(ctx context.Context, backend string, request morpc.Message) (*morpc.Future, error)

type MockRPCAgent 

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

func NewMockLogtailAgent 

func NewMockLogtailAgent() *MockRPCAgent

func (*MockRPCAgent) Close 

func (a *MockRPCAgent) Close()

func (*MockRPCAgent) MockLogtailPRCServerFactory 

func (a *MockRPCAgent) MockLogtailPRCServerFactory(
	name string, address string, logtailServer *service.LogtailServer, options ...morpc.ServerOption) (morpc.RPCServer, error)

func (*MockRPCAgent) MockLogtailRPCClientFactory 

func (a *MockRPCAgent) MockLogtailRPCClientFactory(
	ctx context.Context,
	sid string,
	serverAddr string,
	ownClient morpc.RPCClient,
) (morpc.RPCClient, morpc.Stream, error)

type MockRPCClientStream 

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

func (*MockRPCClientStream) Close 

func (s *MockRPCClientStream) Close(closeConn bool) error

func (*MockRPCClientStream) ID 

func (s *MockRPCClientStream) ID() uint64

func (*MockRPCClientStream) Receive 

func (s *MockRPCClientStream) Receive() (chan morpc.Message, error)

func (*MockRPCClientStream) Send 

func (s *MockRPCClientStream) Send(ctx context.Context, request morpc.Message) error

type PInmemRowsStats 

type PInmemRowsStats struct {
	VisibleCnt                int
	InvisibleCnt              int
	VisibleDistinctBlockCnt   int
	InvisibleDistinctBlockCnt int
}

type PObjectStats 

type PObjectStats struct {
	ObjCnt int
	BlkCnt int
	RowCnt int
}

type PartitionStateStats 

type PartitionStateStats struct {
	DataObjectsVisible   PObjectStats
	DataObjectsInvisible PObjectStats

	TombstoneObjectsVisible   PObjectStats
	TombstoneObjectsInvisible PObjectStats

	InmemRows     PInmemRowsStats
	CheckpointCnt int

	Details struct {
		// 0: locations
		// 1: versions
		DeletedRows    []*batch.Batch
		CheckpointLocs [2][]string
		DataObjectList struct {
			Visible, Invisible []objectio.ObjectEntry
		}
		TombstoneObjectList struct {
			Visible, Invisible []objectio.ObjectEntry
		}
	}
}

func (*PartitionStateStats) String 

func (s *PartitionStateStats) String() string

func (*PartitionStateStats) Summary 

func (s *PartitionStateStats) Summary() PartitionStateStats

type TestDisttaeEngine 

type TestDisttaeEngine struct {
	Engine *disttae.Engine
	// contains filtered or unexported fields
}

func NewTestDisttaeEngine 

func NewTestDisttaeEngine(
	ctx context.Context,
	fs fileservice.FileService,
	rpcAgent *MockRPCAgent,
	storage *TestTxnStorage,
	options ...TestDisttaeEngineOptions,
) (*TestDisttaeEngine, error)

func (*TestDisttaeEngine) Close 

func (de *TestDisttaeEngine) Close(ctx context.Context)

func (*TestDisttaeEngine) CreateDatabaseAndTable 

func (de *TestDisttaeEngine) CreateDatabaseAndTable(
	ctx context.Context,
	databaseName, tableName string,
	schema *catalog2.Schema,
) (
	database engine.Database,
	table engine.Relation,
	err error,
)

func (*TestDisttaeEngine) GetPartitionStateStats 

func (de *TestDisttaeEngine) GetPartitionStateStats(
	ctx context.Context, databaseId, tableId uint64,
) (
	stats PartitionStateStats, err error)

func (*TestDisttaeEngine) GetTable 

func (de *TestDisttaeEngine) GetTable(
	ctx context.Context,
	databaseName, tableName string,
) (
	database engine.Database,
	relation engine.Relation,
	txn client.TxnOperator,
	err error,
)

func (*TestDisttaeEngine) GetTxnClient 

func (de *TestDisttaeEngine) GetTxnClient() client.TxnClient

func (*TestDisttaeEngine) GetTxnOperator 

func (de *TestDisttaeEngine) GetTxnOperator() client.TxnOperator

func (*TestDisttaeEngine) NewTxnOperator 

func (de *TestDisttaeEngine) NewTxnOperator(
	ctx context.Context,
	commitTS timestamp.Timestamp,
	opts ...client.TxnOption,
) (client.TxnOperator, error)

func (*TestDisttaeEngine) Now 

func (de *TestDisttaeEngine) Now() timestamp.Timestamp

func (*TestDisttaeEngine) SubscribeTable 

func (de *TestDisttaeEngine) SubscribeTable(
	ctx context.Context,
	dbID, tbID uint64,
	dbName, tblName string,
	setSubscribed bool,
) (err error)

type TestDisttaeEngineOptions 

type TestDisttaeEngineOptions func(*TestDisttaeEngine)

func WithDisttaeEngineCommitWorkspaceThreshold 

func WithDisttaeEngineCommitWorkspaceThreshold(v uint64) TestDisttaeEngineOptions

func WithDisttaeEngineExtraWorkspaceThreshold 

func WithDisttaeEngineExtraWorkspaceThreshold(v uint64) TestDisttaeEngineOptions

func WithDisttaeEngineInsertEntryMaxCount 

func WithDisttaeEngineInsertEntryMaxCount(v int) TestDisttaeEngineOptions

func WithDisttaeEngineMPool 

func WithDisttaeEngineMPool(mp *mpool.MPool) TestDisttaeEngineOptions

func WithDisttaeEngineQuota 

func WithDisttaeEngineQuota(v uint64) TestDisttaeEngineOptions

func WithDisttaeEngineWriteWorkspaceThreshold 

func WithDisttaeEngineWriteWorkspaceThreshold(v uint64) TestDisttaeEngineOptions

type TestLogtailServer 

type TestLogtailServer struct {
	*service.LogtailServer
}

func NewMockLogtailServer 

func NewMockLogtailServer(
	ctx context.Context, tae *db.DB, cfg *options.LogtailServerCfg, rt runtime.Runtime,
	rpcServerFactory func(string, string, *service.LogtailServer, ...morpc.ServerOption) (morpc.RPCServer, error)) (*TestLogtailServer, error)

type TestOptions 

type TestOptions struct {
	TaeEngineOptions *options.Options
	Timeout          time.Duration
	DisttaeOptions   []TestDisttaeEngineOptions
}

type TestTxnStorage 

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

func NewTestTAEEngine 

func NewTestTAEEngine(
	ctx context.Context, taeDir string, t *testing.T,
	rpcAgent *MockRPCAgent, opts *options.Options) (*TestTxnStorage, error)

func (*TestTxnStorage) BindSchema 

func (ts *TestTxnStorage) BindSchema(schema *catalog.Schema)

func (*TestTxnStorage) Close 

func (ts *TestTxnStorage) Close(destroy bool) error

func (*TestTxnStorage) Commit 

func (ts *TestTxnStorage) Commit(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) CommitTNShard 

func (ts *TestTxnStorage) CommitTNShard(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) Debug 

func (ts *TestTxnStorage) Debug(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) GetDB 

func (ts *TestTxnStorage) GetDB() *db.DB

func (*TestTxnStorage) GetDelegate 

func (ts *TestTxnStorage) GetDelegate() *dbutil.TestEngine

func (*TestTxnStorage) GetRPCHandle 

func (ts *TestTxnStorage) GetRPCHandle() *rpc.Handle

func (*TestTxnStorage) GetStatus 

func (ts *TestTxnStorage) GetStatus(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) Prepare 

func (ts *TestTxnStorage) Prepare(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) Read 

func (ts *TestTxnStorage) Read(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) Rollback 

func (ts *TestTxnStorage) Rollback(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) RollbackTNShard 

func (ts *TestTxnStorage) RollbackTNShard(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) error

func (*TestTxnStorage) Shard 

func (ts *TestTxnStorage) Shard() metadata.TNShard

func (*TestTxnStorage) Start 

func (ts *TestTxnStorage) Start() error

func (*TestTxnStorage) StartTxn 

func (ts *TestTxnStorage) StartTxn() (txnif.AsyncTxn, error)

func (*TestTxnStorage) Write 

func (ts *TestTxnStorage) Write(ctx context.Context, request *txn.TxnRequest, response *txn.TxnResponse) (err error)

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.