compactor

type CompactedIndex interface {
	// IndexProcessor is used for applying custom retention and processing delete requests.
	retention.IndexProcessor
	// Cleanup should clean up all the state built during compaction.
	// It is typically called at the end or in case of an error.
	Cleanup()
	// ToIndexFile is used to convert the CompactedIndex to an IndexFile for uploading to the object store.
	// Once the IndexFile is uploaded using Index.Reader, the file is closed using Index.Close and removed from disk using Index.Path.
	ToIndexFile() (index.Index, error)
}

type Compactor struct {
	services.Service

	DeleteRequestsHandler *deletion.DeleteRequestHandler
	// contains filtered or unexported fields
}

type Config struct {
	WorkingDirectory          string          `yaml:"working_directory"`
	SharedStoreType           string          `yaml:"shared_store"`
	SharedStoreKeyPrefix      string          `yaml:"shared_store_key_prefix"`
	CompactionInterval        time.Duration   `yaml:"compaction_interval"`
	ApplyRetentionInterval    time.Duration   `yaml:"apply_retention_interval"`
	RetentionEnabled          bool            `yaml:"retention_enabled"`
	RetentionDeleteDelay      time.Duration   `yaml:"retention_delete_delay"`
	RetentionDeleteWorkCount  int             `yaml:"retention_delete_worker_count"`
	RetentionTableTimeout     time.Duration   `yaml:"retention_table_timeout"`
	DeleteBatchSize           int             `yaml:"delete_batch_size"`
	DeleteRequestCancelPeriod time.Duration   `yaml:"delete_request_cancel_period"`
	MaxCompactionParallelism  int             `yaml:"max_compaction_parallelism"`
	CompactorRing             util.RingConfig `yaml:"compactor_ring,omitempty"`
	RunOnce                   bool            `yaml:"-"`
}

type IndexCompactor interface {
	// NewTableCompactor returns a new TableCompactor for compacting a table.
	// commonIndexSet refers to common index files or in other words multi-tenant index.
	// existingUserIndexSet refers to existing user specific index files in the storage.
	// makeEmptyUserIndexSetFunc can be used for creating an empty indexSet for a user
	// who does not have an index for it in existingUserIndexSet.
	// periodConfig holds the PeriodConfig for the table.
	NewTableCompactor(
		ctx context.Context,
		commonIndexSet IndexSet,
		existingUserIndexSet map[string]IndexSet,
		makeEmptyUserIndexSetFunc MakeEmptyUserIndexSetFunc,
		periodConfig config.PeriodConfig,
	) TableCompactor

	// OpenCompactedIndexFile opens a compressed index file at given path.
	OpenCompactedIndexFile(
		ctx context.Context,
		path,
		tableName,
		userID,
		workingDir string,
		periodConfig config.PeriodConfig,
		logger log.Logger,
	) (
		CompactedIndex,
		error,
	)
}

type IndexFileConfig struct {
	CompressFile bool
}

type IndexRecords struct {
	Start, NumRecords int
}

type IndexSet interface {
	GetTableName() string
	ListSourceFiles() []storage.IndexFile
	GetSourceFile(indexFile storage.IndexFile) (string, error)
	GetLogger() log.Logger
	GetWorkingDir() string
	// SetCompactedIndex sets the CompactedIndex for upload/applying retention and making the compactor remove the source files.
	// CompactedIndex can be nil only in case of all the source files in common index set being compacted away to per tenant index.
	// It would return an error if the CompactedIndex is nil and removeSourceFiles is true in case of user index set since
	// compaction should either create new files or can be a noop if there is nothing to compact.
	// There is no need to call SetCompactedIndex if no changes were made to the index for this IndexSet.
	SetCompactedIndex(compactedIndex CompactedIndex, removeSourceFiles bool) error
}

type IndexesConfig struct {
	NumUnCompactedFiles, NumCompactedFiles int
}

type MakeEmptyUserIndexSetFunc func(userID string) (IndexSet, error)

type PerUserIndexesConfig struct {
	IndexesConfig
	NumUsers int
}

type TableCompactor interface {
	// CompactTable compacts the table.
	// After compaction is done successfully, it should set the new/updated CompactedIndex for relevant IndexSets.
	CompactTable() (err error)
}