[dbnode] Introduce Aggregator type

src/dbnode/server/options.go

@@ -32,4 +32,5 @@ type StorageOptions struct {
 	TChanNodeServerFn      node.NewTChanNodeServerFn
 	BackgroundProcessFns   []storage.NewBackgroundProcessFn
 	NamespaceHooks         storage.NamespaceHooks
+	NewAggregatorFn        storage.NewAggregatorFn
 }

src/dbnode/server/server.go

@@ -845,6 +845,11 @@ func Run(runOpts RunOptions) {
 		opts = opts.SetNamespaceHooks(runOpts.StorageOptions.NamespaceHooks)
 	}
 
+	if runOpts.StorageOptions.NewAggregatorFn != nil {
+		aggregator := runOpts.StorageOptions.NewAggregatorFn(iopts)
+		opts = opts.SetAggregator(aggregator)
+	}
+
 	// Set bootstrap options - We need to create a topology map provider from the
 	// same topology that will be passed to the cluster so that when we make
 	// bootstrapping decisions they are in sync with the clustered database

src/dbnode/storage/options.go

@@ -174,6 +174,7 @@ type options struct {
 	wideBatchSize                   int
 	newBackgroundProcessFns         []NewBackgroundProcessFn
 	namespaceHooks                  NamespaceHooks
+	aggregator                      Aggregator
 }
 
 // NewOptions creates a new set of storage options with defaults
@@ -249,6 +250,7 @@ func newOptions(poolOpts pool.ObjectPoolOptions) Options {
 		mediatorTickInterval:            defaultMediatorTickInterval,
 		wideBatchSize:                   defaultWideBatchSize,
 		namespaceHooks:                  &noopNamespaceHooks{},
+		aggregator:                      &noopAggregator{},
 	}
 	return o.SetEncodingM3TSZPooled()
 }
@@ -863,6 +865,17 @@ func (o *options) NamespaceHooks() NamespaceHooks {
 	return o.namespaceHooks
 }
 
+func (o *options) SetAggregator(value Aggregator) Options {
+	opts := *o
+	opts.aggregator = value
+
+	return &opts
+}
+
+func (o *options) Aggregator() Aggregator {
+	return o.aggregator
+}
+
 type noOpColdFlush struct{}
 
 func (n *noOpColdFlush) ColdFlushNamespace(Namespace) (OnColdFlushNamespace, error) {
@@ -874,3 +887,15 @@ type noopNamespaceHooks struct{}
 func (h *noopNamespaceHooks) OnCreatedNamespace(Namespace, GetNamespaceFn) error {
 	return nil
 }
+
+type noopAggregator struct{}
+
+func (a *noopAggregator) AggregateTiles(
+	opts AggregateTilesOptions,
+	ns Namespace,
+	shardID uint32,
+	readers []fs.DataFileSetReader,
+	writer fs.StreamingWriter,
+) (int64, error) {
+	return 0, nil
+}

src/dbnode/storage/types.go

@@ -1323,6 +1323,12 @@ type Options interface {
 
 	// NamespaceHooks returns the NamespaceHooks.
 	NamespaceHooks() NamespaceHooks
+
+	// SetAggregator sets the Aggregator.
+	SetAggregator(aggregator Aggregator) Options
+
+	// Aggregator returns the Aggregator.
+	Aggregator() Aggregator
 }
 
 // MemoryTracker tracks memory.
@@ -1394,6 +1400,21 @@ type AggregateTilesOptions struct {
 	InsOptions instrument.Options
 }
 
+// Aggregator is the interface for AggregateTiles.
+type Aggregator interface {
+	// AggregateTiles does tile aggregation.
+	AggregateTiles(
+		opts AggregateTilesOptions,
+		ns Namespace,
+		shardID uint32,
+		readers []fs.DataFileSetReader,
+		writer fs.StreamingWriter,
+	) (int64, error)
+}
+
+// NewAggregatorFn creates a new Aggregator.
+type NewAggregatorFn func(iOpts instrument.Options) Aggregator
+
 // NamespaceHooks allows dynamic plugging into the namespace lifecycle.
 type NamespaceHooks interface {
 	// OnCreatedNamespace gets invoked after each namespace is created.