Add support for materialized views

Author: exAspArk

Dockerfile

@@ -36,7 +36,6 @@ RUN \
 # Compile syncers and server
 
 COPY --chown=app:app src/common/go.mod src/common/go.sum /app/src/common/
-COPY --chown=app:app src/syncer-common/go.mod src/syncer-common/go.sum /app/src/syncer-common/
 
 COPY --chown=app:app src/syncer-postgres/go.mod src/syncer-postgres/go.sum /app/src/syncer-postgres/
 RUN cd /app/src/syncer-postgres && go mod download
@@ -48,7 +47,6 @@ COPY --chown=app:app src/server/go.mod src/server/go.sum /app/src/server/
 RUN cd /app/src/server && go mod download
 
 COPY --chown=app:app src/common /app/src/common
-COPY --chown=app:app src/syncer-common /app/src/syncer-common
 COPY --chown=app:app src/syncer-postgres /app/src/syncer-postgres
 COPY --chown=app:app src/syncer-amplitude /app/src/syncer-amplitude
 COPY --chown=app:app src/server /app/src/server

Dockerfile.test

@@ -56,7 +56,6 @@ RUN \
 # Prepare syncers and server
 
 COPY --chown=app:app src/common/go.mod src/common/go.sum /app/src/common/
-COPY --chown=app:app src/syncer-common/go.mod src/syncer-common/go.sum /app/src/syncer-common/
 
 COPY --chown=app:app src/syncer-postgres/go.mod src/syncer-postgres/go.sum /app/src/syncer-postgres/
 RUN cd /app/src/syncer-postgres && go mod download
@@ -65,7 +64,6 @@ COPY --chown=app:app src/server/go.mod src/server/go.sum /app/src/server/
 RUN cd /app/src/server && go mod download
 
 COPY --chown=app:app src/common /app/src/common
-COPY --chown=app:app src/syncer-common /app/src/syncer-common
 COPY --chown=app:app src/syncer-postgres /app/src/syncer-postgres
 COPY --chown=app:app src/server /app/src/server
 

Makefile

@@ -3,14 +3,12 @@ sh:
 
 install:
 	devbox run "cd src/common && go mod tidy && \
-		cd ../syncer-common && go mod tidy && \
 		cd ../syncer-postgres && go mod tidy && \
 		cd ../syncer-amplitude && go mod tidy && \
 		cd ../server && go mod tidy"
 
 lint:
 	devbox run "cd src/common && go fmt && staticcheck . && \
-		cd ../syncer-common && go fmt && staticcheck . && \
 		cd ../syncer-postgres && go fmt && deadcode . && staticcheck . && \
 		cd ../syncer-amplitude && go fmt && deadcode . && staticcheck . && \
 		cd ../server && go fmt && deadcode . && staticcheck ."

README.md

@@ -323,6 +323,7 @@ SELECT * FROM [TABLE] WHERE [JSON_COLUMN]->>'[JSON_KEY]' = '[JSON_VALUE]';
 - [x] Iceberg tables compaction
 - [x] Packaging in a Docker image
 - [x] Table compaction without Trino as a dependency
+- [x] Materialized views
 - [ ] Partitioned tables ([#15](https://github.com/BemiHQ/BemiDB/issues/15))
 - [ ] Transformations with dbt ([#25](https://github.com/BemiHQ/BemiDB/issues/25))
 

docker/bin-test/test.sh

@@ -28,7 +28,7 @@ SOURCE_POSTGRES_SYNC_MODE=FULL_REFRESH \
   go test -v ./...
 
 # Run tests
-cd /app/src/syncer-common
+cd /app/src/common
 go test -v -count=1 ./...
 cd /app/src/server
 BEMIDB_USER=user \

scripts/catalog.sql

@@ -6,3 +6,11 @@ CREATE TABLE IF NOT EXISTS iceberg_tables (
   previous_metadata_location VARCHAR(1000),
   PRIMARY KEY (catalog_name, table_namespace, table_name)
 );
+
+CREATE TABLE IF NOT EXISTS iceberg_materialized_views (
+  schema_name VARCHAR(255) NOT NULL,
+  table_name VARCHAR(255) NOT NULL,
+  definition TEXT NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_materialized_views ON iceberg_materialized_views (schema_name, table_name);

src/syncer-common/capped_buffer.go src/common/capped_buffer.gosrc/syncer-common/capped_buffer.go renamed to src/common/capped_buffer.go

@@ -1,15 +1,13 @@
-package syncerCommon
+package common
 
 import (
 	"errors"
 	"io"
 	"sync"
-
-	"github.com/BemiHQ/BemiDB/src/common"
 )
 
 type CappedBuffer struct {
-	Config       *common.CommonConfig
+	Config       *CommonConfig
 	MaxSizeBytes int
 
 	buffer          []byte
@@ -20,7 +18,7 @@ type CappedBuffer struct {
 	closed        bool
 }
 
-func NewCappedBuffer(config *common.CommonConfig, maxSizeBytes int) *CappedBuffer {
+func NewCappedBuffer(config *CommonConfig, maxSizeBytes int) *CappedBuffer {
 	sizedBuffer := &CappedBuffer{
 		Config:       config,
 		buffer:       make([]byte, 0, maxSizeBytes),
@@ -44,7 +42,7 @@ func (buf *CappedBuffer) Write(payload []byte) (writtenBytes int, err error) {
 	}
 
 	for len(buf.buffer)+len(payload) > buf.MaxSizeBytes && !buf.closed {
-		common.LogTrace(buf.Config, ">> Waiting for more space in capped buffer...")
+		LogTrace(buf.Config, ">> Waiting for more space in capped buffer...")
 		buf.conditionalSync.Wait() // Wait for the reader
 	}
 
@@ -55,7 +53,7 @@ func (buf *CappedBuffer) Write(payload []byte) (writtenBytes int, err error) {
 
 	writtenBytes = len(payload)
 	buf.buffer = append(buf.buffer, payload...)
-	common.LogTrace(buf.Config, ">> Writing", writtenBytes, "bytes to capped buffer...")
+	LogTrace(buf.Config, ">> Writing", writtenBytes, "bytes to capped buffer...")
 
 	buf.conditionalSync.Broadcast() // Notify the reader that new data is available
 
@@ -72,7 +70,7 @@ func (buf *CappedBuffer) Read(payload []byte) (readBytes int, err error) {
 	defer buf.mutex.Unlock()
 
 	for len(buf.buffer) == 0 && !buf.closed {
-		common.LogTrace(buf.Config, "<< Waiting for more data in capped buffer...")
+		LogTrace(buf.Config, "<< Waiting for more data in capped buffer...")
 		buf.conditionalSync.Wait() // Wait for the writer
 	}
 
@@ -83,7 +81,7 @@ func (buf *CappedBuffer) Read(payload []byte) (readBytes int, err error) {
 	maxReadBytes := len(payload)
 	readBytes = copy(payload, buf.buffer)
 	buf.buffer = buf.buffer[readBytes:]
-	common.LogTrace(buf.Config, "<< Reading "+common.IntToString(readBytes)+"/"+common.IntToString(maxReadBytes)+" bytes from capped buffer...")
+	LogTrace(buf.Config, "<< Reading "+IntToString(readBytes)+"/"+IntToString(maxReadBytes)+" bytes from capped buffer...")
 
 	buf.conditionalSync.Broadcast() // Notify the writer that space is now available
 
@@ -94,7 +92,7 @@ func (buf *CappedBuffer) Close() error {
 	buf.closeOnceSync.Do(func() {
 		buf.mutex.Lock()
 
-		common.LogTrace(buf.Config, "== Closing capped buffer...")
+		LogTrace(buf.Config, "== Closing capped buffer...")
 		buf.closed = true
 
 		buf.conditionalSync.Broadcast() // Wake up any waiting writers/readers

src/syncer-common/capped_buffer_test.go src/common/capped_buffer_test.gosrc/syncer-common/capped_buffer_test.go renamed to src/common/capped_buffer_test.go

@@ -1,18 +1,16 @@
-package syncerCommon
+package common
 
 import (
 	"bytes"
 	"io"
 	"sync"
 	"testing"
 	"time"
-
-	"github.com/BemiHQ/BemiDB/src/common"
 )
 
-func initTestConfig() *common.CommonConfig {
-	return &common.CommonConfig{
-		LogLevel: common.LOG_LEVEL_INFO, // Use INFO to avoid excessive logging during tests
+func initTestConfig() *CommonConfig {
+	return &CommonConfig{
+		LogLevel: LOG_LEVEL_INFO, // Use INFO to avoid excessive logging during tests
 	}
 }
 

src/common/common_config.go

@@ -1,7 +1,7 @@
 package common
 
 const (
-	VERSION = "1.0.1"
+	VERSION = "1.1.0"
 
 	ENV_LOG_LEVEL                   = "BEMIDB_LOG_LEVEL"
 	ENV_DISABLE_ANONYMOUS_ANALYTICS = "BEMIDB_DISABLE_ANONYMOUS_ANALYTICS"

src/common/duckdb_client.go

@@ -8,6 +8,11 @@ import (
 	"github.com/marcboeker/go-duckdb/v2"
 )
 
+var SYNCER_DUCKDB_BOOT_QUERIES = []string{
+	"SET memory_limit='2GB'",
+	"SET threads=2",
+}
+
 type DuckdbClient struct {
 	Config    *CommonConfig
 	Db        *sql.DB
@@ -88,16 +93,21 @@ func (client *DuckdbClient) ExecContext(ctx context.Context, query string, args
 	return client.Db.ExecContext(ctx, replaceNamedStringArgs(query, args[0]))
 }
 
-func (client *DuckdbClient) ExecTransactionContext(ctx context.Context, queries []string) error {
+func (client *DuckdbClient) ExecTransactionContext(ctx context.Context, queries []string, args ...[]map[string]string) error {
 	tx, err := client.Db.Begin()
 	LogDebug(client.Config, "Querying DuckDBClient: BEGIN")
 	if err != nil {
 		return err
 	}
 
-	for _, query := range queries {
+	for i, query := range queries {
 		LogDebug(client.Config, "Querying DuckDBClient:", query)
-		_, err := tx.ExecContext(ctx, query)
+		var err error
+		if len(args) == 0 {
+			_, err = tx.ExecContext(ctx, query)
+		} else {
+			_, err = tx.ExecContext(ctx, replaceNamedStringArgs(query, args[0][i]))
+		}
 		if err != nil {
 			tx.Rollback()
 			return err
@@ -135,7 +145,11 @@ func (client *DuckdbClient) setExplicitAwsCredentials(ctx context.Context) {
 
 func replaceNamedStringArgs(query string, args map[string]string) string {
 	for key, value := range args {
-		query = strings.ReplaceAll(query, "$"+key, value)
+		query = strings.ReplaceAll(
+			query,
+			"$"+key,
+			strings.ReplaceAll(value, "'", "''"),
+		)
 	}
 	return query
 }