memstore

type ArchiveBatch struct {
	common.Batch

	// Size of the batch (number of rows). Notice that compression changes the
	// length of some columns, does not change the size of the batch.
	Size int

	// Version for archive batches.
	Version uint32

	// SeqNum denotes backfill sequence number
	SeqNum uint32

	// For convenience.
	BatchID int32
	Shard   *TableShard
}

type ArchiveJobDetail struct {
	JobDetail

	// Current cutoff.
	CurrentCutoff uint32 `json:"currentCutoff"`
	// Stage of the job is running.
	Stage ArchivingStage `json:"stage"`
	// New Cutoff.
	RunningCutoff uint32 `json:"runningCutoff"`
	// Cutoff of last completed archiving job.
	LastCutoff uint32 `json:"lastCutoff"`
}

type ArchiveJobDetailMutator func(jobDetail *ArchiveJobDetail)

type ArchiveJobDetailReporter func(key string, mutator ArchiveJobDetailMutator)

type ArchiveStore struct {
	// The mutex protects the pointer pointing to the current version of archived vector version.
	sync.RWMutex

	PurgeManager *PurgeManager
	// Current version points to the most recent version of vector store version for queries to use.
	CurrentVersion *ArchiveStoreVersion
}

type ArchiveStoreVersion struct {
	// The mutex
	// protects the Batches map structure and the archiving cutoff field.
	// It does not protect contents within a batch. Before releasing
	// the VectorStore mutex, user should lock the batch level mutex if necessary
	// to ensure proper protection at batch level.
	sync.RWMutex `json:"-"`

	// Wait group used to prevent this ArchiveStore from being evicted
	Users sync.WaitGroup `json:"-"`

	// Each batch in the slice is identified by BaseBatchID+index.
	// Index out of bound and nil Batch for archive batches indicates that none of
	// the columns have been loaded into memory from disk.
	Batches map[int32]*ArchiveBatch `json:"batches"`

	// The archiving cutoff used for this version of the sorted store.
	ArchivingCutoff uint32 `json:"archivingCutoff"`
	// contains filtered or unexported fields
}

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

type ArchivingStage string

type BackfillConfig struct {
	// max buffer size to hold backfill data
	MaxBufferSize int64 `json:"maxBufferSize"`

	// threshold to trigger backfill
	BackfillThresholdInBytes int64 `json:"backfillThresholdInBytes"`
}

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

type BackfillJobDetail struct {
	JobDetail
	// Stage of the job is running.
	Stage BackfillStage `json:"stage"`
	// Current redolog file that's being backfilled.
	RedologFile int64 `json:"redologFile"`
	// Batch offset within the RedologFile.
	BatchOffset uint32 `json:"batchOffset"`
}

type BackfillJobDetailMutator func(jobDetail *BackfillJobDetail)

type BackfillJobDetailReporter func(key string, mutator BackfillJobDetailMutator)

type BackfillManager struct {
	sync.RWMutex `json:"-"`
	BackfillConfig

	// Name of the table.
	TableName string `json:"-"`

	// The shard id of the table.
	Shard int `json:"-"`

	// queue to hold UpsertBatches to backfill
	UpsertBatches []*memCom.UpsertBatch `json:"-"`

	// keep track of the number of records in backfill queue
	NumRecords int `json:"numRecords"`

	// keep track of the size of the buffer that holds batches to be backfilled
	CurrentBufferSize int64 `json:"currentBufferSize"`

	// keep track of the size of the buffer that holds batches being backfilled
	BackfillingBufferSize int64 `json:"backfillingBufferSize"`

	// keep track of the redo log file of the last batch backfilled
	LastRedoFile int64 `json:"lastRedoFile"`

	// keep track of the offset of the last batch backfilled
	LastBatchOffset uint32 `json:"lastBatchOffset"`

	// keep track of the redo log file of the last batch queued
	CurrentRedoFile int64 `json:"currentRedoFile"`

	// keep track of the offset of the last batch being queued
	CurrentBatchOffset uint32 `json:"currentBatchOffset"`

	AppendCond *sync.Cond `json:"-"`
}

type BackfillStage string

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

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

type Job interface {
	JobType() common.JobType
	Run() error
	GetIdentifier() string
	String() string
}

type JobDetail struct {
	// Status of archiving for current table Shard.
	Status JobStatus `json:"status"`

	// Time of next archiving job
	NextRun time.Time `json:"nextRun,omitempty"`

	// Time when the last backfill job finishes
	LastRun time.Time `json:"lastRun,omitempty"`

	// Error of last run if failed.
	LastError error `json:"lastError,omitempty"`
	// Start time of last archiving job.
	LastStartTime time.Time `json:"lastStartTime,omitempty"`
	// Duration of last archiving job.
	LastDuration time.Duration `json:"lastDuration,omitempty"`

	// Number of records processed.
	NumRecords int `json:"numRecords,omitempty"`
	// Number of days affected.
	NumAffectedDays int `json:"numAffectedDays,omitempty"`

	// Total amount of work
	// ie, archiving merge: number of days
	//     archiving snapshot: number of records
	Total int `json:"total,omitempty"`
	// Current finished work.
	Current int `json:"current,omitempty"`

	// Duration for waiting for lock.
	LockDuration time.Duration `json:"lockDuration,omitempty"`
}

type JobStatus string

type LiveBatch struct {
	// The common data structure holding column data.
	common.Batch

	// Capacity of the batch which is decided at the creation time.
	Capacity int

	// maximum of arrival time
	MaxArrivalTime uint32
	// contains filtered or unexported fields
}

type LiveStore struct {
	sync.RWMutex

	// The batch id to batch map.
	Batches map[int32]*LiveBatch

	// Number of rows to create for new batches.
	BatchSize int

	// The upper bound of records (exclusive) that can be read by queries.
	LastReadRecord common.RecordID

	// This is the in memory archiving cutoff time high watermark that gets set by the archiving job
	// before each archiving run. Ingestion will not insert/update records that are older than
	// the archiving cutoff watermark.
	ArchivingCutoffHighWatermark uint32

	// Logs.
	RedoLogManager redolog.RedologManager

	// Manage backfill queue during ingestion.
	BackfillManager *BackfillManager

	// Manage snapshot related stats.
	SnapshotManager *SnapshotManager

	// The writer lock is to guarantee single writer to a Shard at all time. To ensure this, writers
	// (ingestion, archiving etc) need to hold this lock at all times. This lock
	// should be acquired before the VectorStore and Batch locks.
	// TODO: if spinning lock performance is a concern we may need to upgrade this
	// to a designated goroutine with a pc-queue channel.
	WriterLock sync.RWMutex

	// Primary key table of the Shard.
	PrimaryKey common.PrimaryKey

	// The position of the next record to be used for writing. Only used by the ingester.
	NextWriteRecord common.RecordID

	// For convenience.
	HostMemoryManager common.HostMemoryManager `json:"-"`
	// contains filtered or unexported fields
}

type MemStore interface {
	common.TableSchemaReader
	bootstrap.Bootstrapable

	// GetMemoryUsageDetails
	GetMemoryUsageDetails() (map[string]TableShardMemoryUsage, error)
	// GetScheduler returns the scheduler for scheduling archiving and backfill jobs.
	GetScheduler() Scheduler
	// GetHostMemoryManager returns the host memory manager
	GetHostMemoryManager() common.HostMemoryManager
	// AddTableShard add a table shard to the memstore
	AddTableShard(table string, shardID int, needPeerCopy bool)
	// GetTableShard gets the data for a pinned table Shard. Caller needs to unpin after use.
	GetTableShard(table string, shardID int) (*TableShard, error)
	// RemoveTableShard removes table shard from memstore
	RemoveTableShard(table string, shardID int)
	// FetchSchema fetches schema from metaStore and updates in-memory copy of table schema,
	// and set up watch channels for metaStore schema changes, used for bootstrapping mem store.
	FetchSchema() error
	// InitShards loads/recovers data for shards initially owned by the current instance.
	InitShards(schedulerOff bool, shardOwner topology.ShardOwner)
	// HandleIngestion logs an upsert batch and applies it to the in-memory store.
	HandleIngestion(table string, shardID int, upsertBatch *common.UpsertBatch) error
	// Archive is the process moving stable records in fact tables from live batches to archive
	// batches.
	Archive(table string, shardID int, cutoff uint32, reporter ArchiveJobDetailReporter) error

	// Backfill is the process of merging records with event time older than cutoff with
	// archive batches.
	Backfill(table string, shardID int, reporter BackfillJobDetailReporter) error

	// Snapshot is the process to write the current content of dimension table live store in memory to disk.
	Snapshot(table string, shardID int, reporter SnapshotJobDetailReporter) error

	// Purge is the process to purge out of retention archive batches
	Purge(table string, shardID, batchIDStart, batchIDEnd int, reporter PurgeJobDetailReporter) error
}

type Option func(o *Options)

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

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

type PurgeJobDetail struct {
	JobDetail
	// Stage of the job is running.
	Stage PurgeStage `json:"stage"`
	// Number of batches purged
	NumBatches   int `json:"numBatches"`
	BatchIDStart int `json:"batchIDStart"`
	BatchIDEnd   int `json:"batchIDEnd"`
}

type PurgeJobDetailMutator func(jobDetail *PurgeJobDetail)

type PurgeJobDetailReporter func(key string, mutator PurgeJobDetailMutator)

type PurgeManager struct {
	sync.RWMutex `json:"-"`

	// Job Trigger Condition related fields
	// Last purge time.
	LastPurgeTime time.Time `json:"lastPurgeTime"`

	PurgeInterval time.Duration `json:"purgeInterval"`
	// contains filtered or unexported fields
}

type PurgeStage string

type RedoLogBrowser interface {
	ListLogFiles() ([]int64, error)
	ListUpsertBatch(creationTime int64) ([]int64, error)
	ReadData(creationTime int64, upsertBatchOffset int64, start int, length int) (
		[][]interface{}, []string, int, error)
}

type Scheduler interface {
	Start()
	Stop()
	SubmitJob(job Job) (error, chan error)
	DeleteTable(table string, isFactTable bool)
	GetJobDetails(jobType common.JobType) interface{}
	NewBackfillJob(tableName string, shardID int) Job
	NewArchivingJob(tableName string, shardID int, cutoff uint32) Job
	NewSnapshotJob(tableName string, shardID int) Job
	NewPurgeJob(tableName string, shardID int, batchIDStart int, batchIDEnd int) Job
	EnableJobType(jobType common.JobType, enable bool)
	IsJobTypeEnabled(jobType common.JobType) bool
	utils.RWLocker
}

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

type SnapshotJobDetail struct {
	JobDetail
	// Number of mutations in this snapshot.
	NumMutations int `json:"numMutations"`
	// Number of batches written in this snapshot.
	NumBatches int `json:"numBatches"`
	// Current redolog file that's being backfilled.
	RedologFile int64 `json:"redologFile"`
	// Batch offset within the RedologFile.
	BatchOffset uint32 `json:"batchOffset"`
	// Stage of the job is running.
	Stage SnapshotStage `json:"stage"`
}

type SnapshotJobDetailMutator func(jobDetail *SnapshotJobDetail)

type SnapshotJobDetailReporter func(key string, mutator SnapshotJobDetailMutator)

type SnapshotManager struct {
	sync.RWMutex `json:"-"`

	// Number of mutations since last snapshot. Measured as number of rows mutated.
	NumMutations int `json:"numMutations"`

	// Last snapshot time.
	LastSnapshotTime time.Time `json:"'lastSnapshotTime'"`

	// keep track of the redo log file of the last batch snapshotted.
	LastRedoFile int64 `json:"lastRedoFile"`

	// keep track of the offset of the last batch snapshotted
	LastBatchOffset uint32 `json:"lastBatchOffset"`

	// keep track of the record position of the last batch snapshotted
	LastRecord common.RecordID

	// keep track of the redo log file of the last batch queued
	CurrentRedoFile int64 `json:"currentRedoFile"`

	// keep track of the offset of the last batch queued
	CurrentBatchOffset uint32 `json:"currentBatchOffset"`

	// keep track of the record position when last batch queued
	CurrentRecord common.RecordID

	// Configs
	SnapshotInterval time.Duration `json:"snapshotInterval"`

	SnapshotThreshold int `json:"snapshotThreshold"`
	// contains filtered or unexported fields
}

type SnapshotStage string

type TableShard struct {
	// Wait group used to prevent the stores from being prematurely deleted.
	Users sync.WaitGroup `json:"-"`

	ShardID int `json:"-"`

	// For convenience, reference to the table schema struct.
	Schema *common.TableSchema `json:"schema"`

	// Live store. Its locks also cover the primary key.
	LiveStore *LiveStore `json:"liveStore"`

	// Archive store.
	ArchiveStore *ArchiveStore `json:"archiveStore"`

	// For convenience.
	HostMemoryManager common.HostMemoryManager `json:"-"`

	BootstrapState bootstrap.BootstrapState `json:"BootstrapState"`

	// BootstrapDetails shows the details of bootstrap
	BootstrapDetails bootstrap.BootstrapDetails `json:"bootstrapDetails,omitempty"`
	// contains filtered or unexported fields
}

type TableShardMemoryUsage struct {
	ColumnMemory     map[string]*common.ColumnMemoryUsage `json:"cols"`
	PrimaryKeyMemory uint                                 `json:"pk"`
}

type TestFactoryT struct {
	tests.TestFactoryBase
}

type TransferableVectorParty interface {
	// GetHostVectorPartySlice slice vector party between [startIndex, startIndex+length) before transfer to gpu
	GetHostVectorPartySlice(startIndex, length int) common.HostVectorPartySlice
}