Duckdb optimize (#899)

Author: chris-sun-star

build/Dockerfile.sql-analyzer

@@ -3,6 +3,7 @@ FROM golang:1.25 AS builder
 
 ARG GOPROXY
 ARG GOSUMDB
+ARG ASAN_ENABLED
 
 WORKDIR /workspace
 # copy everything
@@ -12,12 +13,20 @@ COPY . .
 RUN if [ -n "$GOPROXY" ]; then go env -w GOPROXY=$GOPROXY; fi
 RUN if [ -n "$GOSUMDB" ]; then go env -w GOSUMDB=$GOSUMDB; fi
 
-RUN make sql-analyzer
+RUN ASAN_ENABLED=$ASAN_ENABLED make sql-analyzer
 
 # start build docker image
 # Use ubuntu:24.04 as the base image
 FROM ubuntu:24.04
 
+ARG ASAN_ENABLED
+# Install libasan if ASAN_ENABLED is set
+RUN if [ -n "$ASAN_ENABLED" ]; then \
+    apt-get update && \
+    apt-get install -y libasan8 && \
+    rm -rf /var/lib/apt/lists/*; \
+    fi
+
 # Set working directory
 WORKDIR /app
 

cmd/sql-analyzer/main.go

@@ -64,7 +64,7 @@ func startHttpServer(ctx context.Context, logger *logrus.Logger) *webserver.HTTP
 	logger.Info("Successfully registered router")
 	go func() {
 		err := httpServer.Run()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			logger.WithError(err).Errorln("Start server failed")
 			os.Exit(1)
 		}
@@ -200,6 +200,17 @@ func main() {
 		}
 	}
 
+	// Configure DuckDB max open conns
+	duckDBMaxOpenConns := 10
+	duckDBMaxOpenConnsStr := os.Getenv("DUCKDB_MAX_OPEN_CONNS")
+	if duckDBMaxOpenConnsStr != "" {
+		if val, err := strconv.Atoi(duckDBMaxOpenConnsStr); err == nil && val > 0 {
+			duckDBMaxOpenConns = val
+		} else {
+			analyzerLogger.Warnf("Invalid DUCKDB_MAX_OPEN_CONNS value '%s', using default of 10.", duckDBMaxOpenConnsStr)
+		}
+	}
+
 	config := &config.Config{
 		Namespace:                    namespace,
 		OBTenant:                     obtenant,
@@ -211,10 +222,11 @@ func main() {
 		QueueSize:                    queueSize,
 		WorkerNum:                    workerNum,
 		PlanCacheSize:                planCacheSize,
+		DuckDBMaxOpenConns:           duckDBMaxOpenConns,
 	}
 
 	// Initialize Stores
-	if err := store.InitGlobalStores(ctx, config.DataPath, collectorLogger); err != nil {
+	if err := store.InitGlobalStores(ctx, config, collectorLogger); err != nil {
 		collectorLogger.Fatalf("Failed to initialize stores: %v", err)
 	}
 

internal/sql-analyzer/config/collector.go

@@ -26,4 +26,5 @@ type Config struct {
 	SqlAuditLimit                int
 	SlowSqlThresholdMilliSeconds int
 	PlanCacheSize                int
+	DuckDBMaxOpenConns           int
 }

internal/sql-analyzer/const/sql/sql_plan.go

@@ -82,20 +82,35 @@ const (
 	GetTablePrimaryKey = `SELECT column_name FROM oceanbase.__all_virtual_column WHERE tenant_id = ? AND table_id = ? and rowkey_position <> 0 ORDER BY rowkey_position`
 	CheckPlanExistence = `SELECT COUNT(*) FROM sql_plan WHERE TENANT_ID = ? AND SVR_IP = ? AND SVR_PORT = ? AND PLAN_ID = ?`
 	GetPlanStats       = `
+		WITH PlanAgg AS (
+			SELECT
+				TENANT_ID,
+				SVR_IP,
+				SVR_PORT,
+				PLAN_ID,
+				PLAN_HASH,
+				MAX(GMT_CREATE) as GMT_CREATE,
+				SUM(IO_COST) as IO_COST,
+				SUM(CPU_COST) as CPU_COST,
+				SUM(COST) as COST,
+				SUM(REAL_COST) as REAL_COST
+			FROM sql_plan
+			WHERE SQL_ID = ?
+			GROUP BY TENANT_ID, SVR_IP, SVR_PORT, PLAN_ID, PLAN_HASH
+		)
 		SELECT
-			TENANT_ID,
-			SVR_IP,
-			SVR_PORT,
-			PLAN_ID,
+			MAX(TENANT_ID) as TENANT_ID,
+			MAX(SVR_IP) as SVR_IP,
+			MAX(SVR_PORT) as SVR_PORT,
+			MAX(PLAN_ID) as PLAN_ID,
 			PLAN_HASH,
-			MIN(GMT_CREATE) as GMT_CREATE,
-			SUM(IO_COST) as IO_COST,
-			SUM(CPU_COST) as CPU_COST,
-			SUM(COST) as COST,
-			SUM(REAL_COST) as REAL_COST
-		FROM sql_plan
-		WHERE SQL_ID = ?
-		GROUP BY TENANT_ID, SVR_IP, SVR_PORT, PLAN_ID, PLAN_HASH
+			MAX(GMT_CREATE) as GMT_CREATE,
+			CAST(AVG(IO_COST) AS BIGINT) as IO_COST,
+			CAST(AVG(CPU_COST) AS BIGINT) as CPU_COST,
+			CAST(AVG(COST) AS BIGINT) as COST,
+			CAST(AVG(REAL_COST) AS BIGINT) as REAL_COST
+		FROM PlanAgg
+		GROUP BY PLAN_HASH
 	`
 	GetTableInfo = `
 		SELECT 

internal/sql-analyzer/store/global.go

@@ -18,21 +18,23 @@ import (
 	"path/filepath"
 
 	"github.com/sirupsen/logrus"
+
+	"github.com/oceanbase/ob-operator/internal/sql-analyzer/config"
 )
 
 var (
 	globalSqlAuditStore *SqlAuditStore
 	globalPlanStore     *PlanStore
 )
 
-func InitGlobalStores(ctx context.Context, dataPath string, logger *logrus.Logger) error {
+func InitGlobalStores(ctx context.Context, conf *config.Config, logger *logrus.Logger) error {
 	var err error
-	globalSqlAuditStore, err = NewSqlAuditStore(ctx, filepath.Join(dataPath, "sql_audit"), logger)
+	globalSqlAuditStore, err = NewSqlAuditStore(ctx, filepath.Join(conf.DataPath, "sql_audit"), conf.DuckDBMaxOpenConns, logger)
 	if err != nil {
 		return err
 	}
 
-	globalPlanStore, err = NewPlanStore(ctx, filepath.Join(dataPath, "sql_plan"), logger)
+	globalPlanStore, err = NewPlanStore(ctx, filepath.Join(conf.DataPath, "sql_plan"), conf.DuckDBMaxOpenConns, logger)
 	if err != nil {
 		return err
 	}

internal/sql-analyzer/store/sql_audit.go

@@ -44,13 +44,15 @@ type SqlAuditStore struct {
 	Logger *logger.Logger
 }
 
-func NewSqlAuditStore(c context.Context, path string, l *logger.Logger) (*SqlAuditStore, error) {
+func NewSqlAuditStore(c context.Context, path string, maxOpenConns int, l *logger.Logger) (*SqlAuditStore, error) {
 	// Use an in-memory DuckDB database for operations.
 	db, err := sql.Open("duckdb", "") // In-memory
 	if err != nil {
 		return nil, fmt.Errorf("failed to open in-memory duckdb: %w", err)
 	}
 
+	db.SetMaxOpenConns(maxOpenConns)
+
 	// Set memory limit for in-memory DB
 	conn, err := db.Conn(c)
 	if err == nil {

internal/sql-analyzer/store/sql_plan.go

@@ -42,7 +42,7 @@ func (s *PlanStore) InitSqlPlanTable() error {
 	return err
 }
 
-func NewPlanStore(c context.Context, path string, l *logger.Logger) (*PlanStore, error) {
+func NewPlanStore(c context.Context, path string, maxOpenConns int, l *logger.Logger) (*PlanStore, error) {
 	l.Infof("Using plan store at %s", path)
 	if err := os.MkdirAll(path, 0755); err != nil {
 		return nil, fmt.Errorf("failed to create data directory %s: %w", path, err)
@@ -77,6 +77,8 @@ func NewPlanStore(c context.Context, path string, l *logger.Logger) (*PlanStore,
 		return nil, errors.Wrapf(err, "failed to open duckdb after retries at path %s", path)
 	}
 
+	db.SetMaxOpenConns(maxOpenConns)
+
 	// Set memory limit
 	memLimit := os.Getenv("DUCKDB_MEMORY_LIMIT")
 	if memLimit == "" {

make/sql-analyzer.mk

@@ -15,10 +15,16 @@ sql-analyzer-doc-gen: sql-analyzer-dep-install ## Generate swagger docs for sql-
 sql-analyzer: sql-analyzer-doc-gen
 	@echo Building sql-analyzer...
 	@mkdir -p bin
+ifneq ($(ASAN_ENABLED),)
+	@echo "Building with AddressSanitizer..."
+	CGO_ENABLED=1 CGO_CFLAGS="-fsanitize=address -g" CGO_LDFLAGS="-fsanitize=address" go build -tags dynamic -o bin/sql-analyzer cmd/sql-analyzer/main.go
+else
 	@go build -o bin/sql-analyzer cmd/sql-analyzer/main.go
+endif
 
 .PHONY: sql-analyzer-image
 sql-analyzer-image:
 	$(eval DOCKER_BUILD_ARGS :=)
 	$(if $(GOPROXY),$(eval DOCKER_BUILD_ARGS := --build-arg GOPROXY=$(GOPROXY)))
+	$(if $(ASAN_ENABLED),$(eval DOCKER_BUILD_ARGS := $(DOCKER_BUILD_ARGS) --build-arg ASAN_ENABLED=$(ASAN_ENABLED)))
 	docker build $(DOCKER_BUILD_ARGS) -t ${SQL_ANALYZER_IMG} -f build/Dockerfile.sql-analyzer .