[ENH]: refactoring log service (#1958)

## Description of changes
The goal was to refactor the log service to make it simpler and use
simpler abstraction on top of postgres. A next pr in the stack is adding
model/property testing.

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
	 - Create a new log service with update offset position

## Test plan
Not tested for now in this branch, another pr is adding property
testing.

Author: nicolasgere

.gitattributes

@@ -1,2 +1,4 @@
 *_pb2.py* linguist-generated
 *_pb2_grpc.py* linguist-generated
+go/database/**/db/** linguist-generated=true
+go/pkg/proto/** linguist-generated=true

.gitignore

@@ -38,3 +38,6 @@ target/
 
 # environment file generated by the Javascript tests
 .chroma_env
+
+# Rapid test data
+testdata

go/Makefile

@@ -16,3 +16,17 @@ clean:
 
 docker:
 	docker build -t chroma-coordinator:latest .
+
+
+DATABABASE_LOG_DIR := database/log
+
+log_db_clean:
+	rm -rf $(DATABABASE_LOG_DIR)/db
+
+.PHONY: quota_db_generate
+log_db_generate: log_db_clean
+	sqlc generate -f $(DATABABASE_LOG_DIR)/sqlc.yaml
+
+log_db_migration:
+	 atlas migrate diff initial --to file://$(DATABABASE_LOG_DIR)/schema --dev-url "docker://postgres/15/dev"  --format '{{ sql . "  " }}' --dir file://$(DATABABASE_LOG_DIR)/migrations
+

go/database/log/atlas.hcl

@@ -0,0 +1,13 @@
+
+env "dev" {
+  url = "postgresql://chroma:[email protected]:5432/log?sslmode=disable"
+  migration {
+    dir = "file://migrations"
+  }
+}
+env "prod" {
+  url = getenv("DB_URL")
+  migration {
+    dir = "file://migrations"
+  }
+}

go/database/log/db/copyfrom.go

@@ -0,0 +1,15 @@
+-- Create "collection" table
+CREATE TABLE "public"."collection" (
+  "id" text NOT NULL,
+  "record_compaction_offset_position" bigint NOT NULL,
+  "record_enumeration_offset_position" bigint NOT NULL,
+  PRIMARY KEY ("id")
+);
+-- Create "record_log" table
+CREATE TABLE "public"."record_log" (
+  "offset" bigint NOT NULL,
+  "collection_id" text NOT NULL,
+  "timestamp" integer NOT NULL DEFAULT (EXTRACT(epoch FROM now()))::integer,
+  "record" bytea NOT NULL,
+  PRIMARY KEY ("collection_id", "offset")
+);

go/database/log/db/db.go

@@ -0,0 +1,2 @@
+h1:l718NRul/xO5Vz4nKzlWAR9ML+kOkn4TTgIlMQYcUZA=
+20240401221053_initial.sql h1:RPywT3bZIeCHgfStvajW3fcDhqadDY5xI9MFjE/Un4U=

go/database/log/db/models.go

@@ -0,0 +1,31 @@
+-- name: GetCollectionForUpdate :one
+SELECT *
+FROM collection
+WHERE id = $1
+FOR UPDATE;
+
+-- name: InsertRecord :copyfrom
+INSERT INTO record_log (collection_id, "offset", record) values($1, $2, $3);
+
+-- name: GetRecordsForCollection :many
+SELECT * FROM record_log r WHERE r.collection_id = $1 AND r.offset > $2  ORDER BY r.offset DESC limit $3 ;
+
+-- name: GetAllCollectionsToCompact :many
+with summary as (
+    select r.collection_id, r.offset, r.timestamp, row_number() over(partition by r.collection_id order by r.offset) as rank
+    from record_log r, collection c
+    where r.collection_id = c.id
+    and r.offset > c.record_compaction_offset_position
+)
+select * from summary
+where rank=1
+order by timestamp;
+
+-- name: UpdateCollectionCompactionOffsetPosition :exec
+UPDATE collection set record_compaction_offset_position = $2 where id = $1;
+
+-- name: UpdateCollectionEnumerationOffsetPosition :exec
+UPDATE collection set record_enumeration_offset_position = $2 where id = $1;
+
+-- name: InsertCollection :one
+INSERT INTO collection (id, record_enumeration_offset_position, record_compaction_offset_position) values($1, $2, $3) returning *;

go/database/log/db/queries.sql.go

@@ -0,0 +1,8 @@
+CREATE TABLE collection (
+                        id text PRIMARY KEY,
+                        record_compaction_offset_position bigint NOT NULL,
+                        record_enumeration_offset_position bigint NOT NULL
+                        );
+
+-- The `record_compaction_offset_position` column indicates the offset position of the latest compaction.
+-- The `record_enenumeration_offset_position` column denotes the incremental offset for the most recent record in a collection.

go/database/log/migrations/20240401221053_initial.sql

@@ -0,0 +1,8 @@
+CREATE TABLE record_log (
+                        "offset" BIGINT NOT NULL,
+                        collection_id text NOT NULL,
+                        timestamp int NOT NULL default extract(epoch from now())::int,
+                        record bytea NOT NULL,
+                        PRIMARY KEY(collection_id, "offset")
+);
+

go/database/log/migrations/atlas.sum

@@ -0,0 +1,10 @@
+version: "2"
+sql:
+  - engine: "postgresql"
+    queries: "queries/"
+    schema: "schema/"
+    gen:
+      go:
+        package: "log"
+        out: "db"
+        sql_package: "pgx/v5"