Generalized merging function

[kelindar]

This PR generalizes Merge() function and removes now redundant Add() functions for numbers. When Merge() function is called on numerical columns without specifying a custom merge option on the column, the behavior of Add() will be observed (i.e. increments by default).

Example

In this example we are creating a column called location that contains a JSON-encoded position and velocity. The position is updated by calling a MergeString() function with the velocity vector, the updates are then merged atomically using a merging function specified.

The merge happens when transaction is committed to ensure consistency. Hence, this technique allows for two concurrent transactions to update the same position.

// Movement represents a movement with a position and velocity
type Movement struct {
    Position [2]float64 `json:"position,omitempty"`
    Velocity [2]float64 `json:"velocity,omitempty"`
}

func main() {

    // A merging function that accepts a velocity vector and updates
    // the movement structure accordingly.
    mergeVectors := func(value, delta string) string {
        movement, ok := parseMovement(value)
        if !ok {
            movement = Movement{
                Position: [2]float64{0, 0},
            }
        }

        // Parse the incoming delta value
        velocity, ok := parseVector(delta)
        if !ok {
            return value
        }

        // Update the current velocity and recalculate the position
        movement.Velocity = velocity
        movement.Position[0] += velocity[0] // Update X
        movement.Position[1] += velocity[1] // Update Y

        // Encode the movement as JSON and return the updated value
        return encode(movement)
    }

    // Create a column with a specified merge function
    db := column.NewCollection()
    db.CreateColumn("location", column.ForString(
        column.WithMerge(mergeVectors), // use our merging function
    ))

    // Insert an empty row
    id, _ := db.Insert(func(r column.Row) error {
        r.SetString("location", "{}")
        return nil
    })

    // Update 20 times
    for i := 0; i < 20; i++ {

        // Move the location by applying a same velocity vector
        db.Query(func(txn *column.Txn) error {
            location := txn.String("location")
            return txn.QueryAt(id, func(r column.Row) error {
                location.Merge(encode([2]float64{1, 2}))
                return nil
            })
        })

        // Print out current location
        db.Query(func(txn *column.Txn) error {
            location := txn.String("location")
            return txn.QueryAt(id, func(r column.Row) error {
                value, _ := location.Get()
                fmt.Printf("%.2d: %v \n", i, value)
                return nil
            })
        })
    }

}

// parseMovement parses a value string into a Movement struct
func parseMovement(value string) (out Movement, ok bool) {
    if err := json.Unmarshal([]byte(value), &out); err != nil {
        return Movement{}, false
    }
    return out, true
}

// parseVector parses a value string into 2 dimensional array
func parseVector(value string) (out [2]float64, ok bool) {
    if err := json.Unmarshal([]byte(value), &out); err != nil {
        return [2]float64{}, false
    }
    return out, true
}

// encodes encodes the value as JSON
func encode(value any) string {
    encoded, _ := json.Marshal(value)
    return string(encoded)
}