Speed up full-refresh syncing and write to Iceberg w/o Trino as a dependency

Author: exAspArk

img/architecture.png

@@ -8,7 +8,7 @@ import (
 )
 
 const (
-	VERSION = "1.0.0-beta.1"
+	VERSION = "1.0.0-beta.2"
 
 	ENV_PORT     = "BEMIDB_PORT"
 	ENV_DATABASE = "BEMIDB_DATABASE"

src/server/config.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"database/sql"
-	"regexp"
 	"slices"
 	"strings"
 
@@ -26,6 +25,7 @@ var DUCKDB_INIT_BOOT_QUERIES = []string{
 	// Configure DuckDB
 	"SET scalar_subquery_error_on_multiple_rows=false",
 	"SET timezone='UTC'",
+	"SET memory_limit='2GB'",
 }
 
 type Duckdb struct {
@@ -146,10 +146,8 @@ func (duckdb *Duckdb) setExplicitAwsCredentials(ctx context.Context) {
 }
 
 func replaceNamedStringArgs(query string, args map[string]string) string {
-	re := regexp.MustCompile(`['";]`) // Escape single quotes, double quotes, and semicolons from args
-
 	for key, value := range args {
-		query = strings.ReplaceAll(query, "$"+key, re.ReplaceAllString(value, ""))
+		query = strings.ReplaceAll(query, "$"+key, value)
 	}
 	return query
 }

src/server/duckdb.go

@@ -537,7 +537,7 @@ func TestHandleQuery(t *testing.T) {
 			"SELECT float4_column FROM postgres.test_table WHERE float4_column != 3.14": {
 				"description": {"float4_column"},
 				"types":       {uint32ToString(pgtype.Float4OID)},
-				"values":      {"NaN"},
+				"values":      {"0"},
 			},
 			"SELECT float8_column FROM postgres.test_table WHERE bool_column = TRUE": {
 				"description": {"float8_column"},
@@ -590,12 +590,12 @@ func TestHandleQuery(t *testing.T) {
 				"values":      {"12:00:00.123"},
 			},
 			"SELECT timeMsColumn FROM postgres.test_table WHERE timeMsColumn IS NOT NULL": {
-				"description": {"timemscolumn"},
+				"description": {"timeMsColumn"},
 				"types":       {uint32ToString(pgtype.TimeOID)},
 				"values":      {"12:00:00.123"},
 			},
 			"SELECT timeMsColumn FROM postgres.test_table WHERE timeMsColumn IS NULL": {
-				"description": {"timemscolumn"},
+				"description": {"timeMsColumn"},
 				"types":       {uint32ToString(pgtype.TimeOID)},
 				"values":      {""},
 			},
@@ -672,7 +672,7 @@ func TestHandleQuery(t *testing.T) {
 			"SELECT interval_column FROM postgres.test_table WHERE interval_column IS NOT NULL": {
 				"description": {"interval_column"},
 				"types":       {uint32ToString(pgtype.TextOID)},
-				"values":      {"2806201000001us"},
+				"values":      {"1 mon 2 days 01:00:01.000001"},
 			},
 			"SELECT interval_column FROM postgres.test_table WHERE interval_column IS NULL": {
 				"description": {"interval_column"},
@@ -737,7 +737,7 @@ func TestHandleQuery(t *testing.T) {
 			"SELECT point_column FROM postgres.test_table WHERE point_column IS NOT NULL": {
 				"description": {"point_column"},
 				"types":       {uint32ToString(pgtype.TextOID)},
-				"values":      {"0101000000D6FFFF5F74AC4240CDFFFFDF44804640"},
+				"values":      {"(37.347301483154,45.002101898193)"},
 			},
 			"SELECT point_column FROM postgres.test_table WHERE point_column IS NULL": {
 				"description": {"point_column"},

src/server/query_handler_test.go

@@ -25,49 +25,49 @@ type MetadataJson struct {
 }
 
 type StorageS3 struct {
-	s3Client *s3.Client
-	config   *Config
+	S3Client *s3.Client
+	Config   *Config
 }
 
-func NewS3Storage(config *Config) *StorageS3 {
+func NewS3Storage(Config *Config) *StorageS3 {
 	var awsConfigOptions = []func(*awsConfig.LoadOptions) error{
-		awsConfig.WithRegion(config.Aws.Region),
+		awsConfig.WithRegion(Config.Aws.Region),
 	}
 
-	if config.LogLevel == LOG_LEVEL_TRACE {
+	if Config.LogLevel == LOG_LEVEL_TRACE {
 		awsConfigOptions = append(awsConfigOptions, awsConfig.WithClientLogMode(aws.LogRequest))
 	}
 
-	if IsLocalHost(config.Aws.S3Endpoint) {
-		awsConfigOptions = append(awsConfigOptions, awsConfig.WithBaseEndpoint("http://"+config.Aws.S3Endpoint))
+	if IsLocalHost(Config.Aws.S3Endpoint) {
+		awsConfigOptions = append(awsConfigOptions, awsConfig.WithBaseEndpoint("http://"+Config.Aws.S3Endpoint))
 	} else {
-		awsConfigOptions = append(awsConfigOptions, awsConfig.WithBaseEndpoint("https://"+config.Aws.S3Endpoint))
+		awsConfigOptions = append(awsConfigOptions, awsConfig.WithBaseEndpoint("https://"+Config.Aws.S3Endpoint))
 	}
 
 	awsCredentials := credentials.NewStaticCredentialsProvider(
-		config.Aws.AccessKeyId,
-		config.Aws.SecretAccessKey,
+		Config.Aws.AccessKeyId,
+		Config.Aws.SecretAccessKey,
 		"",
 	)
 	awsConfigOptions = append(awsConfigOptions, awsConfig.WithCredentialsProvider(awsCredentials))
 
 	loadedAwsConfig, err := awsConfig.LoadDefaultConfig(context.Background(), awsConfigOptions...)
-	PanicIfError(config, err)
+	PanicIfError(Config, err)
 
-	s3Client := s3.NewFromConfig(loadedAwsConfig, func(o *s3.Options) {
-		if config.Aws.S3Endpoint != DEFAULT_AWS_S3_ENDPOINT {
+	S3Client := s3.NewFromConfig(loadedAwsConfig, func(o *s3.Options) {
+		if Config.Aws.S3Endpoint != DEFAULT_AWS_S3_ENDPOINT {
 			o.UsePathStyle = true
 		}
 	})
 
 	return &StorageS3{
-		s3Client: s3Client,
-		config:   config,
+		S3Client: S3Client,
+		Config:   Config,
 	}
 }
 
 func (storage *StorageS3) IcebergTableFields(metadataPath string) ([]IcebergTableField, error) {
-	metadataKey := strings.TrimPrefix(metadataPath, "s3://"+storage.config.Aws.S3Bucket+"/")
+	metadataKey := strings.TrimPrefix(metadataPath, "s3://"+storage.Config.Aws.S3Bucket+"/")
 	metadataContent, err := storage.readFileContent(metadataKey)
 	if err != nil {
 		return nil, err
@@ -78,8 +78,8 @@ func (storage *StorageS3) IcebergTableFields(metadataPath string) ([]IcebergTabl
 
 func (storage *StorageS3) readFileContent(fileKey string) ([]byte, error) {
 	ctx := context.Background()
-	getObjectResponse, err := storage.s3Client.GetObject(ctx, &s3.GetObjectInput{
-		Bucket: aws.String(storage.config.Aws.S3Bucket),
+	getObjectResponse, err := storage.S3Client.GetObject(ctx, &s3.GetObjectInput{
+		Bucket: aws.String(storage.Config.Aws.S3Bucket),
 		Key:    aws.String(fileKey),
 	})
 	if err != nil {

src/server/storage_s3.go

@@ -40,11 +40,15 @@ func registerFlags() {
 	_config.BaseConfig = &common.BaseConfig{}
 
 	flag.StringVar(&_config.BaseConfig.LogLevel, "log-level", os.Getenv(common.ENV_LOG_LEVEL), `Log level: "ERROR", "WARN", "INFO", "DEBUG", "TRACE". Default: "`+common.DEFAULT_LOG_LEVEL+`"`)
+	flag.StringVar(&_config.BaseConfig.CatalogDatabaseUrl, "catalog-database-url", os.Getenv(common.ENV_CATALOG_DATABASE_URL), "Catalog database URL")
 	flag.StringVar(&_config.BaseConfig.DestinationSchemaName, "destination-schema-name", os.Getenv(common.ENV_DESTINATION_SCHEMA_NAME), "Destination schema name to store the synced data")
 	flag.StringVar(&_config.BaseConfig.Trino.DatabaseUrl, "trino-database-url", os.Getenv(common.ENV_TRINO_DATABASE_URL), "Trino database URL to sync to")
 	flag.StringVar(&_config.BaseConfig.Trino.CatalogName, "trino-catalog-name", os.Getenv(common.ENV_TRINO_CATALOG_NAME), "Trino catalog name")
+	flag.StringVar(&_config.BaseConfig.Aws.Region, "aws-region", os.Getenv(common.ENV_AWS_REGION), "AWS region")
 	flag.StringVar(&_config.BaseConfig.Aws.S3Endpoint, "aws-s3-endpoint", os.Getenv(common.ENV_AWS_S3_ENDPOINT), "AWS S3 endpoint. Default: \""+common.DEFAULT_AWS_S3_ENDPOINT+`"`)
 	flag.StringVar(&_config.BaseConfig.Aws.S3Bucket, "aws-s3-bucket", os.Getenv(common.ENV_AWS_S3_BUCKET), "AWS S3 bucket name")
+	flag.StringVar(&_config.BaseConfig.Aws.AccessKeyId, "aws-access-key-id", os.Getenv(common.ENV_AWS_ACCESS_KEY_ID), "AWS access key ID")
+	flag.StringVar(&_config.BaseConfig.Aws.SecretAccessKey, "aws-secret-access-key", os.Getenv(common.ENV_AWS_SECRET_ACCESS_KEY), "AWS secret access key")
 	flag.BoolVar(&_config.BaseConfig.DisableAnonymousAnalytics, "disable-anonymous-analytics", os.Getenv(common.ENV_DISABLE_ANONYMOUS_ANALYTICS) == "true", "Disable anonymous analytics collection")
 
 	flag.StringVar(&_config.ApiKey, "api-key", os.Getenv(ENV_API_KEY), "Amplitude API Key")
@@ -60,6 +64,9 @@ func parseFlags() {
 	} else if !slices.Contains(common.LOG_LEVELS, _config.BaseConfig.LogLevel) {
 		panic("Invalid log level " + _config.BaseConfig.LogLevel + ". Must be one of " + strings.Join(common.LOG_LEVELS, ", "))
 	}
+	if _config.BaseConfig.CatalogDatabaseUrl == "" {
+		panic("Catalog database URL is required")
+	}
 	if _config.BaseConfig.DestinationSchemaName == "" {
 		panic("Destination schema name is required")
 	}
@@ -69,6 +76,22 @@ func parseFlags() {
 	if _config.BaseConfig.Trino.CatalogName == "" {
 		panic("Trino catalog name is required")
 	}
+	if _config.BaseConfig.Aws.Region == "" {
+		panic("AWS region is required")
+	}
+	if _config.BaseConfig.Aws.S3Endpoint == "" {
+		_config.BaseConfig.Aws.S3Endpoint = common.DEFAULT_AWS_S3_ENDPOINT
+	}
+	if _config.BaseConfig.Aws.S3Bucket == "" {
+		panic("AWS S3 bucket name is required")
+	}
+	if _config.BaseConfig.Aws.AccessKeyId != "" && _config.BaseConfig.Aws.SecretAccessKey == "" {
+		panic("AWS secret access key is required")
+	}
+	if _config.BaseConfig.Aws.AccessKeyId == "" && _config.BaseConfig.Aws.SecretAccessKey != "" {
+		panic("AWS access key ID is required")
+	}
+
 	if _config.ApiKey == "" {
 		panic("Amplitude API key is required")
 	}

src/syncer-amplitude/config.go

@@ -7,14 +7,67 @@ require github.com/BemiHQ/BemiDB/src/syncer-common v0.0.1
 replace github.com/BemiHQ/BemiDB/src/syncer-common => ../syncer-common
 
 require (
+	github.com/apache/arrow-go/v18 v18.1.0 // indirect
+	github.com/apache/arrow/go/arrow v0.0.0-20200730104253-651201b0f516 // indirect
+	github.com/apache/thrift v0.21.0 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.36.6 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.11 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.29.18 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.71 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.33 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.85 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.37 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.37 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.37 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.7.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.18 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.18 // indirect
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.84.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.25.6 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.34.1 // indirect
+	github.com/aws/smithy-go v1.22.4 // indirect
+	github.com/duckdb/duckdb-go-bindings v0.1.17 // indirect
+	github.com/duckdb/duckdb-go-bindings/darwin-amd64 v0.1.12 // indirect
+	github.com/duckdb/duckdb-go-bindings/darwin-arm64 v0.1.12 // indirect
+	github.com/duckdb/duckdb-go-bindings/linux-amd64 v0.1.12 // indirect
+	github.com/duckdb/duckdb-go-bindings/linux-arm64 v0.1.12 // indirect
+	github.com/duckdb/duckdb-go-bindings/windows-amd64 v0.1.12 // indirect
+	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
+	github.com/goccy/go-json v0.10.5 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/flatbuffers v25.1.24+incompatible // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/pgx/v5 v5.7.5 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
 	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/goidentity/v6 v6.0.1 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
+	github.com/linkedin/goavro v2.1.0+incompatible // indirect
+	github.com/marcboeker/go-duckdb/arrowmapping v0.0.10 // indirect
+	github.com/marcboeker/go-duckdb/mapping v0.0.11 // indirect
+	github.com/marcboeker/go-duckdb/v2 v2.3.3 // indirect
+	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/trinodb/trino-go-client v0.323.0 // indirect
-	golang.org/x/crypto v0.37.0 // indirect
-	golang.org/x/net v0.24.0 // indirect
+	github.com/xitongsys/parquet-go v1.6.2 // indirect
+	github.com/xitongsys/parquet-go-source v0.0.0-20241021075129-b732d2ac9c9b // indirect
+	github.com/zeebo/xxh3 v1.0.2 // indirect
+	golang.org/x/crypto v0.40.0 // indirect
+	golang.org/x/exp v0.0.0-20250718183923-645b1fa84792 // indirect
+	golang.org/x/mod v0.26.0 // indirect
+	golang.org/x/net v0.42.0 // indirect
+	golang.org/x/sync v0.16.0 // indirect
+	golang.org/x/sys v0.34.0 // indirect
+	golang.org/x/text v0.27.0 // indirect
+	golang.org/x/tools v0.35.0 // indirect
+	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 )