Log the number of currently active connections

Author: exAspArk

src/common/duckdb_client.go

@@ -67,25 +67,25 @@ func NewDuckdbClient(config *CommonConfig, bootQueries ...[]string) *DuckdbClien
 }
 
 func (client *DuckdbClient) QueryContext(ctx context.Context, query string) (*sql.Rows, error) {
-	LogDebug(client.Config, "Querying DuckDBClient:", query)
+	LogDebug(client.Config, "Querying DuckDB:", query)
 	return client.Db.QueryContext(ctx, query)
 }
 
 func (client *DuckdbClient) QueryRowContext(ctx context.Context, query string, args ...map[string]string) *sql.Row {
-	LogDebug(client.Config, "Querying DuckDBClient:", query)
+	LogDebug(client.Config, "Querying DuckDB row:", query)
 	if len(args) == 0 {
 		return client.Db.QueryRowContext(ctx, query)
 	}
 	return client.Db.QueryRowContext(ctx, replaceNamedStringArgs(query, args[0]))
 }
 
 func (client *DuckdbClient) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
-	LogDebug(client.Config, "Preparing DuckDBClient statement:", query)
+	LogDebug(client.Config, "Preparing DuckDB statement:", query)
 	return client.Db.PrepareContext(ctx, query)
 }
 
 func (client *DuckdbClient) ExecContext(ctx context.Context, query string, args ...map[string]string) (sql.Result, error) {
-	LogDebug(client.Config, "Querying DuckDBClient:", query)
+	LogDebug(client.Config, "Executing DuckDB:", query)
 	if len(args) == 0 {
 		return client.Db.ExecContext(ctx, query)
 	}
@@ -95,13 +95,13 @@ func (client *DuckdbClient) ExecContext(ctx context.Context, query string, args
 
 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")
+	LogDebug(client.Config, "Executing DuckDB: BEGIN")
 	if err != nil {
 		return err
 	}
 
 	for i, query := range queries {
-		LogDebug(client.Config, "Querying DuckDBClient:", query)
+		LogDebug(client.Config, "Executing DuckDB in transaction:", query)
 		var err error
 		if len(args) == 0 {
 			_, err = tx.ExecContext(ctx, query)
@@ -114,7 +114,7 @@ func (client *DuckdbClient) ExecTransactionContext(ctx context.Context, queries
 		}
 	}
 
-	LogDebug(client.Config, "Querying DuckDBClient: COMMIT")
+	LogDebug(client.Config, "Executing DuckDB: COMMIT")
 	return tx.Commit()
 }
 

src/common/iceberg_table_writer.go

@@ -129,9 +129,6 @@ func (writer *IcebergTableWriter) UniqueIndexColumnNames() []string {
 			uniqueIndexColumnNames = append(uniqueIndexColumnNames, icebergSchemaColumn.ColumnName)
 		}
 	}
-	if len(uniqueIndexColumnNames) == 0 {
-		Panic(writer.Config, "Cannot update Iceberg table without unique index columns: "+writer.IcebergTable.String())
-	}
 	return uniqueIndexColumnNames
 }
 
@@ -597,7 +594,7 @@ func (writer *IcebergTableWriter) deleteObject(key string) {
 		fileName = strings.Split(key, "/metadata/")[1]
 	}
 
-	LogInfo(writer.Config, "Deleting object:", fileName)
+	LogDebug(writer.Config, "Deleting object:", fileName)
 	writer.StorageS3.S3Client.DeleteObject(key)
 }
 

src/server/main.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	_ "net/http/pprof"
 	"slices"
+	"sync/atomic"
 
 	"github.com/BemiHQ/BemiDB/src/common"
 )
@@ -33,15 +34,18 @@ func main() {
 
 	queryHandler := NewQueryHandler(config, duckdbClient)
 
+	var connectionCount int64 = 0
 	for {
 		conn := AcceptConnection(config, tcpListener)
-		common.LogInfo(config.CommonConfig, "BemiDB: Accepted connection from", conn.RemoteAddr())
+		atomic.AddInt64(&connectionCount, 1)
+		common.LogInfo(config.CommonConfig, "BemiDB: Accepted", common.Int64ToString(atomic.LoadInt64(&connectionCount))+"th", "connection from", conn.RemoteAddr())
 		server := NewPostgresServer(config, &conn)
 
 		go func() {
 			server.Run(queryHandler)
 			defer server.Close()
-			common.LogInfo(config.CommonConfig, "BemiDB: Closed connection from", conn.RemoteAddr())
+			common.LogInfo(config.CommonConfig, "BemiDB: Closed", common.Int64ToString(atomic.LoadInt64(&connectionCount))+"th", "connection from", conn.RemoteAddr())
+			atomic.AddInt64(&connectionCount, -1)
 		}()
 	}
 }

src/syncer-postgres/lib/debezium.go

@@ -18,6 +18,7 @@ type DebeziumRowSource struct {
 	Schema        string `json:"schema"`
 	Table         string `json:"table"`
 	CommittedAtNs int64  `json:"ts_ns"`
+	Position      int64  `json:"lsn"`
 }
 
 type DebeziumRowData struct {