Remove Aggregation op from applied pipeline and add Equal() to variou…

…s operations and pipeline

op/applied/type.go

@@ -36,29 +36,40 @@ type Rollup struct {
 	AggregationID aggregation.ID
 }
 
+// Equal determines whether two rollup operations are equal.
+func (op Rollup) Equal(other Rollup) bool {
+	return op.AggregationID == other.AggregationID && bytes.Equal(op.ID, other.ID)
+}
+
 func (op Rollup) String() string {
-	var b bytes.Buffer
-	b.WriteString("{")
-	fmt.Fprintf(&b, "id: %s, ", op.ID)
-	fmt.Fprintf(&b, "aggregation: %v", op.AggregationID)
-	b.WriteString("}")
-	return b.String()
+	return fmt.Sprintf("{id: %s, aggregation: %v}", op.ID, op.AggregationID)
 }
 
 // Union is a union of different types of operation.
 type Union struct {
 	Type           op.Type
-	Aggregation    op.Aggregation
 	Transformation op.Transformation
 	Rollup         Rollup
 }
 
+// Equal determines whether two operation unions are equal.
+func (u Union) Equal(other Union) bool {
+	if u.Type != other.Type {
+		return false
+	}
+	switch u.Type {
+	case op.TransformationType:
+		return u.Transformation.Equal(other.Transformation)
+	case op.RollupType:
+		return u.Rollup.Equal(other.Rollup)
+	}
+	return true
+}
+
 func (u Union) String() string {
 	var b bytes.Buffer
 	b.WriteString("{")
 	switch u.Type {
-	case op.AggregationType:
-		fmt.Fprintf(&b, "aggregation: %s", u.Aggregation.String())
 	case op.TransformationType:
 		fmt.Fprintf(&b, "transformation: %s", u.Transformation.String())
 	case op.RollupType:
@@ -81,6 +92,19 @@ func (p Pipeline) IsEmpty() bool {
 	return len(p.Operations) == 0
 }
 
+// Equal determines whether two pipelines are equal.
+func (p Pipeline) Equal(other Pipeline) bool {
+	if len(p.Operations) != len(other.Operations) {
+		return false
+	}
+	for i := 0; i < len(p.Operations); i++ {
+		if !p.Operations[i].Equal(other.Operations[i]) {
+			return false
+		}
+	}
+	return true
+}
+
 func (p Pipeline) String() string {
 	var b bytes.Buffer
 	b.WriteString("{operations: [")

op/type.go

@@ -45,6 +45,11 @@ type Aggregation struct {
 	Type aggregation.Type
 }
 
+// Equal determines whether two aggregation operations are equal.
+func (op Aggregation) Equal(other Aggregation) bool {
+	return op.Type == other.Type
+}
+
 func (op Aggregation) String() string {
 	return op.Type.String()
 }
@@ -55,6 +60,11 @@ type Transformation struct {
 	Type transformation.Type
 }
 
+// Equal determines whether two transformation operations are equal.
+func (op Transformation) Equal(other Transformation) bool {
+	return op.Type == other.Type
+}
+
 func (op Transformation) String() string {
 	return op.Type.String()
 }