init

Author: tompston

.gitignore

@@ -0,0 +1,2 @@
+
+.DS_Store

README.md

@@ -0,0 +1,32 @@
+## Time-series upsert performance & compression
+
+A small check to benchmark the performance of upsert queries for time-series data in:
+
+- mongodb
+- postgresql
+- postgresql with timescaledb extension
+
+**NOTE** that this was written in a hurry and the end results should be taken with a grain of salt.
+
+### Setup
+
+- mongodb on localhost:27017, no auth
+- postgresql on localhost:5432, user: postgres, password: postgres, db: test
+- install timescaledb extension on postgresql
+
+### Result
+
+```
+go test -benchmem -run=^$ -bench ^BenchmarkDbs$ ts-benchmark
+
+BenchmarkDbs/MongoDB-compound-single-upsert-10         	       1	1394460875 ns/op	76327976 B/op	 1140562 allocs/op
+BenchmarkDbs/Postgres-native-single-10                 	       1	1376155583 ns/op	 2488368 B/op	   90038 allocs/op
+BenchmarkDbs/Postgres-timescale-single_uspert-10       	       1	1849769834 ns/op	 2480400 B/op	   89993 allocs/op
+BenchmarkDbs/Postgres-timescale-batch_upsert-10        	       3	 412889889 ns/op	11025413 B/op	   90033 allocs/op
+
+Table size summary for 10000 records:
+------------------------
+MongoDB: 			1308 kb
+Postgres (native): 		2168 kb
+Postgres (timescale): 		24 kb
+```

benchmark_test.go

@@ -0,0 +1,144 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"testing"
+
+	"ts-benchmark/dbs/data"
+	"ts-benchmark/dbs/mongodb"
+	"ts-benchmark/dbs/postgres"
+
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+const (
+	TABLE = "timeseries"
+)
+
+func BenchmarkDbs(b *testing.B) {
+
+	pgdb, err := postgres.NewConn("localhost", 5432, "postgres", "postgres", "test")
+	if err != nil {
+		panic(err)
+	}
+	defer pgdb.Close(context.Background())
+
+	const NUM_RECORDS = 10000
+
+	// Generate fake data which is reused for both MongoDB and Postgres
+	fake := data.CreateFakeData(NUM_RECORDS)
+
+	var kbMongo, kbPostgresNative, kbPostgresTimescale int32
+
+	b.Run("MongoDB-compound-single-upsert", func(b *testing.B) {
+		db, err := mongodb.NewConn("localhost", 27017, "", "")
+		if err != nil {
+			b.Fatalf("Failed to connect to MongoDB: %v", err)
+		}
+		defer db.Disconnect(context.Background())
+
+		coll := db.Database("test").Collection(TABLE)
+		if err := mongodb.Setup(coll); err != nil {
+			b.Fatalf("Failed to setup MongoDB: %v", err)
+		}
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			t, err := mongodb.Upsert(coll, fake)
+			if err != nil {
+				b.Fatalf("Failed to Upsert in MongoDB: %v", err)
+			}
+			_ = t
+		}
+		b.StopTimer()
+
+		// Get the statistics of the collection
+		var stats bson.M
+		command := bson.D{{Key: "collStats", Value: "timeseries"}}
+		err = db.Database("test").RunCommand(context.TODO(), command).Decode(&stats)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		bytes, ok := stats["totalSize"].(int32)
+		if !ok {
+			log.Fatal("Failed to convert to int32")
+		}
+
+		kbMongo = bytes / 1024
+	})
+
+	b.Run("Postgres-native-single", func(b *testing.B) {
+
+		if err := postgres.SetupVanilla(pgdb); err != nil {
+			b.Fatalf("Failed to setup Postgres: %v", err)
+		}
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			t, err := postgres.UpsertVanilla(pgdb, fake)
+			if err != nil {
+				b.Fatalf("Failed to Upsert in Postgres: %v", err)
+			}
+			_ = t
+		}
+		b.StopTimer()
+
+		tableSize, err := postgres.GetTableSize(pgdb, "timeseries")
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		kbPostgresNative = int32(tableSize)
+	})
+
+	b.Run("Postgres-timescale-single uspert", func(b *testing.B) {
+		if err := postgres.SetupTimescaleDB(pgdb); err != nil {
+			b.Fatalf("Failed to setup Postgres: %v", err)
+		}
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			t, err := postgres.UpsertTimescaleDB(pgdb, fake)
+			if err != nil {
+				b.Fatalf("Failed to Upsert in Postgres: %v", err)
+			}
+
+			_ = t
+		}
+		b.StopTimer()
+
+		tableSize, err := postgres.GetTableSize(pgdb, "timeseries_timescale")
+		if err != nil {
+			log.Fatal(err)
+		}
+		kbPostgresTimescale = int32(tableSize)
+	})
+
+	b.Run("Postgres-timescale-batch upsert", func(b *testing.B) {
+		if err := postgres.SetupTimescaleDB(pgdb); err != nil {
+			b.Fatalf("Failed to setup Postgres: %v", err)
+		}
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			t, err := postgres.UpsertBatchTimescaleDB(pgdb, fake)
+			if err != nil {
+				b.Fatalf("Failed to Upsert in Postgres: %v", err)
+			}
+			_ = t
+		}
+		b.StopTimer()
+
+	})
+
+	fmt.Printf(`
+Table size summary for %v records:
+------------------------
+MongoDB: 			%v kb
+Postgres (native): 		%v kb
+Postgres (timescale): 		%v kb
+`, NUM_RECORDS, kbMongo, kbPostgresNative, kbPostgresTimescale)
+}

dbs/data/main.go

@@ -0,0 +1,32 @@
+package data
+
+import (
+	"math"
+	"time"
+)
+
+type Timeseries struct {
+	StartTime time.Time `json:"start_time" bson:"start_time"`
+	Interval  int       `json:"interval" bson:"interval"`
+	Area      string    `json:"area" bson:"area"`
+	Value     float64   `json:"value" bson:"value"`
+}
+
+func CreateFakeData(numRows int) []Timeseries {
+	var docs []Timeseries
+
+	for i := 0; i < numRows; i++ {
+		// random value
+		val := math.Round(math.Sin(float64(i)) * 100)
+
+		doc := Timeseries{
+			StartTime: time.Now().Add(time.Duration(i) * time.Hour).Truncate(time.Minute),
+			Interval:  36000000,
+			Area:      "area1",
+			Value:     val,
+		}
+		docs = append(docs, doc)
+	}
+
+	return docs
+}

dbs/influxdb/main.go

@@ -0,0 +1,60 @@
+package influxdb
+
+/*
+
+brew install influxdb
+influxd
+
+brew services start influxdb
+
+*/
+
+import (
+	"context"
+	"fmt"
+	"time"
+	"ts-benchmark/dbs/data"
+
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+)
+
+func NewConn(url, token string) influxdb2.Client {
+	client := influxdb2.NewClient(url, token)
+	return client
+}
+
+func Setup(client influxdb2.Client, org, bucket string) error {
+	deleteAPI := client.DeleteAPI()
+
+	// Define the time range for deletion (from the beginning of time to now)
+	start := time.Unix(0, 0)
+	end := time.Now().AddDate(2, 0, 0)
+
+	// Delete previous data
+	err := deleteAPI.DeleteWithName(context.Background(), org, bucket, start, end, `_measurement="timeseries"`)
+	if err != nil {
+		return fmt.Errorf("failed to delete previous data: %v", err)
+	}
+
+	return nil
+}
+
+func Upsert(client influxdb2.Client, org, bucket string, docs []data.Timeseries) error {
+	writeAPI := client.WriteAPIBlocking(org, bucket)
+
+	// Upsert new data
+	for _, doc := range docs {
+		p := influxdb2.NewPointWithMeasurement("timeseries").
+			// AddTag("interval", fmt.Sprintf("%d", doc.Interval)).
+			AddTag("area", doc.Area).
+			AddField("value", doc.Value).
+			SetTime(doc.StartTime)
+
+		err := writeAPI.WritePoint(context.Background(), p)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

dbs/mongodb/main.go

@@ -0,0 +1,87 @@
+package mongodb
+
+import (
+	"context"
+	"fmt"
+	"time"
+	"ts-benchmark/dbs/data"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func NewConn(host string, port int, username, password string) (*mongo.Client, error) {
+	opt := options.Client().
+		SetMaxPoolSize(20).                  // Set the maximum number of connections in the connection pool
+		SetMaxConnIdleTime(10 * time.Minute) // Close idle connections after the specified time
+
+	// If both the username and password exists, use it as the credentials. Else use the non-authenticated url.
+	var url string
+	if username != "" && password != "" {
+		opt.SetAuth(options.Credential{Username: username, Password: password})
+		url = fmt.Sprintf("mongodb://%s:%s@%s:%d", username, password, host, port)
+	} else {
+		url = fmt.Sprintf("mongodb://%s:%d", host, port)
+	}
+
+	opt.ApplyURI(url)
+
+	conn, err := mongo.Connect(context.Background(), opt)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := conn.Ping(context.Background(), nil); err != nil {
+		return nil, fmt.Errorf("failed to ping mongodb: %v", err)
+	}
+
+	return conn, nil
+}
+
+func Setup(coll *mongo.Collection) error {
+	// Delete previous data
+	if _, err := coll.DeleteMany(context.Background(), bson.M{}); err != nil {
+		return err
+	}
+
+	// Create index on the "start_time", "interval" and "area" fields
+	indexModel := mongo.IndexModel{
+		Keys: bson.D{
+			{Key: "start_time", Value: -1},
+			{Key: "interval", Value: -1},
+			{Key: "area", Value: -1},
+		},
+	}
+
+	if _, err := coll.Indexes().CreateOne(context.Background(), indexModel); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func Upsert(coll *mongo.Collection, docs []data.Timeseries) (time.Duration, error) {
+	now := time.Now()
+
+	for _, doc := range docs {
+		filter := map[string]interface{}{
+			"start_time": doc.StartTime,
+			"interval":   doc.Interval,
+			"area":       doc.Area,
+		}
+
+		update := map[string]interface{}{
+			"$set": map[string]interface{}{
+				"value": doc.Value,
+			},
+		}
+
+		_, err := coll.UpdateOne(context.Background(), filter, update, options.Update().SetUpsert(true))
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	return time.Since(now), nil
+}