sif

type Accumulator interface {
	Accumulate(row Row) error                  // Accumulate adds a row to this Accumulator
	Merge(o Accumulator) error                 // Merge merges another Accumulator into this one
	ToBytes() ([]byte, error)                  // ToBytes serializes this Accumulator
	FromBytes(buf []byte) (Accumulator, error) // FromBytes produces a *new* Accumulator from serialized data
}

type AccumulatorFactory interface {
	New() Accumulator
}

type BucketedPartitionIndex interface {
	PartitionIndex
	GetBucket(bucket uint64) PartitionIndex // return the PartitionIndex associated with the given bucket
}

type BuildablePartition interface {
	Partition
	ForEachRow(fn MapOperation) error            // ForEachRow iterates over Rows in a Partition
	CanInsertRowData(row Row) error              // CanInsertRowData checks if a Row can be inserted into this Partition
	AppendEmptyRowData(tempRow Row) (Row, error) // AppendEmptyRowData is a convenient way to add an empty Row to the end of this Partition, returning the Row so that Row methods can be used to populate it
	AppendRowData(row Row) error                 // AppendRowData adds a Row to the end of this Partition, if it isn't full and if the Row matches this Partition's schema
	InsertRowData(row Row, pos int) error        // InsertRowData inserts a Row at a specific position within this Partition, if it isn't full and if the Row matches this Partition's Schema. Other Rows are shifted as necessary.
	TruncateRowData(numRows int)                 // TruncateRowData zeroes out rows from the current last row towards the beginning of the Partition
	CreateTempRow() Row
}

type CollectedPartition interface {
	Partition
	ForEachRow(fn MapOperation) error // ForEachRow iterates over Rows in a Partition
}

type Column interface {
	Clone() Column            // Clone returns a copy of this Column
	Index() int               // Index returns the index of this Column within a Schema
	Type() ColumnType         // Type returns the ColumnType of this Column
	Accessor() ColumnAccessor // Accessor returns a ColumnAccessor for this Column
}

type ColumnAccessor interface {
	Name() string
	String() string
	Rename(newName string)
	Clone() ColumnAccessor
	Type() ColumnType
	IsNil(row Row) bool   // IsNil returns true iff the given column value is nil in this row. If an error occurs, this function will return false.
	SetNil(row Row) error // SetNil sets the given column value to nil within this row
}

type ColumnType interface {
	ToString(v interface{}) string           // produces a string representation of a value of this Type
	Deserialize([]byte) (interface{}, error) // Defines how this type is deserialized, returning an error if a value of the correct type could not be deserialized from the given bytes
}

type Compressor interface {
	Compress(source io.Reader, dest io.Writer) error   // Compress compresseses data from a read stream to a write stream
	Decompress(source io.Reader, dest io.Writer) error // Decompress decompresses data from a read stream to a write stream
	Destroy()                                          // Destroy cleans up anything relevant when the Compressor is no longer needed
}

type DataFrame interface {
	GetSchema() Schema                            // GetSchema returns the schema of the data at this task.
	GetDataSource() DataSource                    // GetDataSource returns the DataSource of a DataFrame
	GetParser() DataSourceParser                  // GetParser returns the DataSourceParser of a DataFrame
	To(...*DataFrameOperation) (DataFrame, error) // To is a "functional operations" factory method for DataFrames, chaining operations onto the current one(s).
}

type DataFrameOperation struct {
	TaskType TaskType // the task type
	Do       func(df DataFrame) (*DataFrameOperationResult, error)
}

type DataFrameOperationResult struct {
	Task       Task   // the task
	DataSchema Schema // the new data schema
}

type DataSource interface {
	Analyze() (PartitionMap, error)
	DeserializeLoader([]byte) (PartitionLoader, error)
	IsStreaming() bool
}

type DataSourceParser interface {
	PartitionSize() int // returns the maximum size of Partitions produced by this DataSourceParser, in rows
	Parse(
		r io.Reader,
		source DataSource,
		schema Schema,
		onIteratorEnd func(),
	) (PartitionIterator, error) // lazily converts bytes from a Reader into Partitions
}

type FilterOperation func(row Row) (bool, error)

type FixedWidthColumnType interface {
	ColumnType
	Size() int                                       // returns size in bytes of a column type
	SerializeFixed(v interface{}, dest []byte) error // Similar to Serialize, but accepts a reusable buffer of guaranteed Size()
}

type FlatMapOperation func(row Row, newRow RowFactory) error

type GenericAccumulator[T any] interface {
	Accumulator
	Value() T // Value returns the accumulated value of this Accumulator
}

type GenericAccumulatorFactory[T any] interface {
	AccumulatorFactory
	Value(Accumulator) (T, error) // retrieves the value from the given accumulator or an error if the given accumulator does not match the expected type
}

type GenericColumnAccessor[T any] interface {
	ColumnAccessor
	CloneT() GenericColumnAccessor[T]
	TypeT() GenericColumnType[T]
	From(row Row) (T, error)
	To(row Row, val T) error
}

type GenericColumnType[T any] interface {
	ColumnType
	ToStringT(v T) string           // produces a string representation of a value of this Type
	DeserializeT([]byte) (T, error) // Defines how this type is deserialized, returning an error if a value of the correct type could not be deserialized from the given bytes
}

type GenericFixedWidthColumnType[T any] interface {
	GenericColumnType[T]
	FixedWidthColumnType
	SerializeFixedT(v T, dest []byte) error // Similar to SerializeT, but accepts a reusable buffer of guaranteed Size()
}

type GenericVariableWidthColumnType[T any] interface {
	GenericColumnType[T]
	VariableWidthColumnType
	SerializeT(v T) ([]byte, error) // Similar to SerializeT, but accepts a reusable buffer of guaranteed Size()
}

type KeyablePartition interface {
	KeyRows(kfn KeyingOperation) (OperablePartition, error) // KeyRows generates hash keys for a row from a key column. Attempts to manipulate partition in-place, falling back to creating a fresh partition if there are row errors
	IsKeyed() bool                                          // IsKeyed returns true iff this Partition has been keyed with KeyRows
	GetKey(rowNum int) (uint64, error)                      // GetKey returns the key for a particular row number, or returns an error if the Partition is not keyed.
	GetRowKey(row Row) (uint64, error)                      // GetRowKey returns the key for a particular row, or returns an error if the Partition is not keyed.
	GetKeyRange(start int, end int) ([]uint64, error)       // GetKeyRange returns keys for the given range of rows, or returns an error if the Partition is not keyed.
}

type KeyingOperation func(row Row) ([]byte, error)

type MapOperation func(row Row) error

type Numeric interface {
	int8 | int16 | int32 | int64 | uint8 | uint16 | uint32 | uint64 | float32 | float64
}

type OperablePartition interface {
	Partition
	KeyablePartition
	UpdateSchema(currentSchema Schema)                                                  // UpdateSchema sets the public schema of a Partition
	AddColumn(accessor ColumnAccessor) (OperablePartition, error)                       // AddColumn adds space for a new column to this Partition
	RemoveColumn(colName string) (OperablePartition, error)                             // RemoveColumn removes a column's data and metadata from this Partition
	RenameColumn(colName string, newAccessor ColumnAccessor) (OperablePartition, error) // RenameColumn modifies the name for a column in Partition data
	MapRows(fn MapOperation) (OperablePartition, error)                                 // MapRows runs a MapOperation on each row in this Partition, manipulating them in-place. Will fall back to creating a fresh partition if PartitionRowErrors occur.
	FlatMapRows(fn FlatMapOperation) ([]OperablePartition, error)                       // FlatMapRows runs a FlatMapOperation on each row in this Partition, creating new Partitions
	FilterRows(fn FilterOperation) (OperablePartition, error)                           // FilterRows filters the Rows in the current Partition, creating a new one. Rows are retained if FilterOperation returns true
}

type Partition interface {
	ID() string            // ID retrieves the ID of this Partition
	GetMaxRows() int       // GetMaxRows retrieves the maximum number of rows in this Partition
	GetNumRows() int       // GetNumRows retrieves the number of rows in this Partition
	GetRow(rowNum int) Row // GetRow retrieves a specific row from this Partition
	GetSchema() Schema     // GetSchema retrieves the schema for this Partition
}

type PartitionAssignmentStrategy interface {
}

type PartitionCache interface {
	Destroy()
	Add(key string, value ReduceablePartition)
	Get(key string) (value ReduceablePartition, err error) // removes the partition from the cache and returns it, if present. Returns an error otherwise.
	GetSerialized(key string) ([]byte, error)              // removes the partition from the cache in a serialized format, if present. Returns an error otherwise.
	CurrentSize() int
	Resize(frac float64) bool // resize by a fraction RELATIVE TO THE CURRENT NUMBER OF ITEMS IN THE CACHE
}

type PartitionIndex interface {
	SetMaxRows(maxRows int)                                                      // Change the maxRows for future partitions created by this index
	GetNextStageSchema() Schema                                                  // Returns the Schema for the Stage which will *read* from this index
	MergePartition(part ReduceablePartition, reducefn ReductionOperation) error  // Merges all the Rows within a keyed Partition into this PartitionIndex. reducefn may be nil, indicating that reduction is not intended.
	MergeRow(tempRow Row, row Row, reducefn ReductionOperation) error            // Merges a keyed Row of data into the PartitionIndex, possibly appending it to an existing/new Partition, or combining it with an existing Row. reducefn may be nil, indicating that reduction is not intended.
	GetPartitionIterator(destructive bool) PartitionIterator                     // Returns the PartitionIterator for this PartitionIndex. Must always return the same PartitionIterator, even if called multiple times.
	GetSerializedPartitionIterator(destructive bool) SerializedPartitionIterator // Returns a SerializedPartitionIterator for this PartitionIndex. Must always return the same SerializedPartitionIterator, even if called multiple times.
	NumPartitions() uint64                                                       // Returns the number of Partitions in this PartitionIndex
	CacheSize() int                                                              // Returns the in-memory size (in Partitions) of the underlying PartitionCache
	ResizeCache(frac float64) bool                                               // Resizes the underlying PartitionCache
	Destroy()                                                                    // Destroys the index
}

type PartitionIterator interface {
	HasNextPartition() bool
	// if unlockPartition is not nil, it must be called when one is finished with the returned Partition
	NextPartition() (part Partition, unlockPartition func(), err error)
	OnEnd(onEnd func())
}

type PartitionLoader interface {
	ToString() string                                        // for logging
	Load(parser DataSourceParser) (PartitionIterator, error) // how to actually load data
	GobEncode() ([]byte, error)                              // how to serialize this PartitionLoader
	GobDecode([]byte) error                                  // how to deserialize this PartitionLoader
}

type PartitionMap interface {
	HasNext() bool
	Next() PartitionLoader
}

type ReduceablePartition interface {
	BuildablePartition
	KeyablePartition
	SerializablePartition
	PopulateTempRow(tempRow Row, idx int)
	FindFirstKey(key uint64) (int, error)                                          // PRECONDITION: Partition must already be sorted by key
	FindLastKey(key uint64) (int, error)                                           // PRECONDITION: Partition must already be sorted by key
	FindFirstRowKey(keyBuf []byte, key uint64, keyfn KeyingOperation) (int, error) // PRECONDITION: Partition must already be sorted by key
	FindLastRowKey(keyBuf []byte, key uint64, keyfn KeyingOperation) (int, error)  // PRECONDITION: Partition must already be sorted by key
	AverageKeyValue() (uint64, error)                                              // AverageKeyValue is the average value of key within this sorted, keyed Partition
	Split(pos int) (ReduceablePartition, ReduceablePartition, error)               // Split splits a Partition into two Partitions. Split position ends up in right Partition.
	BalancedSplit() (uint64, ReduceablePartition, ReduceablePartition, error)      // Split position ends up in right Partition.
}

type ReductionOperation func(lrow Row, rrow Row) error

type Row interface {
	GetPartID() string                               // Returns the ID of the containing partition (if there is one)
	GetKey() (uint64, error)                         // Returns the key for this Row, or an error if the Row's Partition isn't keyed
	Schema() Schema                                  // Schema returns a read-only copy of the schema for a row
	String() string                                  // String returns a string representation of this row
	IsNil(colName string) bool                       // IsNil returns true iff the given column value is nil in this row. If an error occurs, this function will return false.
	SetNil(colName string) error                     // SetNil sets the given column value to nil within this row
	Get(colName string) (val interface{}, err error) // Get returns the value of any column as an interface{}, if it exists
	Set(colName string, val interface{}) error       // Set sets the value of any column to an interface{} (or nil), if it exists
}

type RowFactory func() Row

type RuntimeStatistics interface {
	// GetStartTime returns the start time of the Sif pipeline
	GetStartTime() time.Time
	// GetRuntime returns the running time of the Sif pipeline
	GetRuntime() time.Duration
	// GetNumRowsProcessed returns the number of Rows which have been processed so far, counted by stage
	GetNumRowsProcessed() []int64
	// GetNumPartitionsProcessed returns the number of Partitions which have been processed so far, counted by stage
	GetNumPartitionsProcessed() []int64
	// GetCurrentPartitionProcessingTime returns a rolling average of partition processing time
	GetCurrentPartitionProcessingTime() time.Duration
	// GetStageRuntimes returns all recorded stage runtimes, from the most recent run of each Stage
	GetStageRuntimes() []time.Duration
	// GetStageTransformRuntimes returns all recorded stage transform-phase runtimes, from the most recent run of each Stage
	GetStageTransformRuntimes() []time.Duration
	// GetStageShuffleRuntimes returns all recorded stage shuffle-phase runtimes, from the most recent run of each Stage
	GetStageShuffleRuntimes() []time.Duration
}

type Schema interface {
	Equals(otherSchema Schema) error                            // Equals return true iff this Schema is identical to otherSchema
	Clone() Schema                                              // Clone() returns a deep copy of this Schema
	NumColumns() int                                            // NumColumns returns the number of columns in this Schema.
	NumFixedWidthColumns() int                                  // NumFixedWidthColumns returns the number of fixed-length columns in this Schema.
	NumVariableWidthColumns() int                               // NumVariableWidthColumns returns the number of variable-length columns in this Schema.
	GetColumn(colName string) (col Column, err error)           // GetColumn returns the Column associated with colName, or an error if none exists
	HasColumn(colName string) bool                              // HasColumn returns true iff colName corresponds to a Column in this Schema
	ColumnNames() []string                                      // ColumnNames returns a list of Column names in this Schema, in the order they were created
	ColumnTypes() []ColumnType                                  // ColumnTypes returns a list of Column types in this Schema, in the order they were created
	ColumnAccessors() []ColumnAccessor                          // ColumnAccessors returns a list of ColumnAccessors in this Schema, in the order they were created
	ForEachColumn(fn func(name string, col Column) error) error // ForEachColumn runs a function for each Column in this Schema
}

type SerializablePartition interface {
	Partition
	SetCompressor(Compressor) // SetCompressor defines the compressor for this Partition. Must be called prior to serializing.
	ToBytes() ([]byte, error) // ToBytes serializes a Partition to a byte array suitable for persistence to disk
}

type SerializedPartitionIterator interface {
	HasNextSerializedPartition() bool
	NextSerializedPartition() (id string, spart []byte, done func(), err error)
	OnEnd(onEnd func())
}

type StageContext interface {
	context.Context
	ShuffleBuckets() []uint64                                // ShuffleBuckets returns the shuffle buckets for this stage, or an empty slice if there are none
	SetShuffleBuckets([]uint64) error                        // SetShuffleBuckets configures the shuffle buckets for this Stage
	OutgoingSchema() Schema                                  // OutgoingSchema returns the initial underlying data schema for the next stage (or the current Stage if this is the last Stage)
	SetOutgoingSchema(schema Schema) error                   // SetOutgoingSchema sets the initial underlying data schema for the next stage within this StageContext (or sets to the current Stage schema if this is the last Stage)
	PartitionCache() PartitionCache                          // PartitionCache returns the configured PartitionCache for this Stage, or nil if none exists
	SetPartitionCache(cache PartitionCache) error            // SetPartitionCache configures the PartitionCache for this Stage, returning an error if one is already set
	PartitionIndex() PartitionIndex                          // PartitionIndex returns the PartitionIndex for this StageContext, or nil if one has not been set
	SetPartitionIndex(idx PartitionIndex) error              // SetPartitionIndex sets the PartitionIndex for this StageContext. An error is returned if one has already been set.
	IncomingPartitionIterator() PartitionIterator            // IncomingPartitionIndex returns the incoming PartitionIterator for this StageContext, or nil if one has not been set
	SetIncomingPartitionIterator(i PartitionIterator) error  // SetIncomingPartitionIndex sets the incoming PartitionIterator for this StageContext. An error is returned if one has already been set.
	KeyingOperation() KeyingOperation                        // KeyingOperation retrieves the KeyingOperation for this Stage (if it exists)
	SetKeyingOperation(keyFn KeyingOperation) error          // Configure the keying operation for the end of this stage
	ReductionOperation() ReductionOperation                  // ReductionOperation retrieves the ReductionOperation for this Stage (if it exists)
	SetReductionOperation(reduceFn ReductionOperation) error // Configure the reduction operation for the end of this stage
	Accumulator() Accumulator                                // Accumulator retrieves the Accumulator for this Stage (if it exists)
	SetAccumulator(acc Accumulator) error                    // Configure the accumulator for the end of this stage
	CollectionLimit() int                                    // CollectionLimit retrieves the CollectionLimit for this Stage (or -1 if unset)
	SetCollectionLimit(limit int) error                      // Configure the CollectionLimit for the end of this stage
	TargetPartitionSize() int                                // TargetPartitionSize returns the intended Partition maxSize for outgoing Partitions
	SetTargetPartitionSize(TargetPartitionSize int) error    // SetTargetPartitionSize configures the intended Partition maxSize for outgoing Partitions
	Destroy() error                                          // Destroys anything using a lot of memory or goroutines within this StageContext
}

type Task interface {
	RunInitialize(sctx StageContext) error
	RunWorker(sctx StageContext, previous OperablePartition) ([]OperablePartition, error)
	RunCoordinator(sctx StageContext) error
}

type TaskType string

type VariableWidthColumnType interface {
	ColumnType
	Serialize(v interface{}) ([]byte, error) // Defines how this type is serialized, returning an error if the given value is not of the expected type, or if something went wrong during serialization
}