Merge pull request #1686 from cloudquery/fix/profile_events_memory

fix: Don't scan profile events if listener is not set

Author: kavirajk

conn_process.go

@@ -1,4 +1,3 @@
-
 package clickhouse
 
 import (
@@ -185,11 +184,14 @@ func (c *connect) handle(ctx context.Context, packet byte, on *onProcess) error
 		}
 		c.debugf("[table columns]")
 	case proto.ServerProfileEvents:
-		events, err := c.profileEvents(ctx)
+		scanEvents := on.profileEvents != nil
+		events, err := c.profileEvents(ctx, scanEvents)
 		if err != nil {
 			return err
 		}
-		on.profileEvents(events)
+		if scanEvents {
+			on.profileEvents(events)
+		}
 	case proto.ServerLog:
 		logs, err := c.logs(ctx)
 		if err != nil {

conn_profile_events.go

@@ -1,4 +1,3 @@
-
 package clickhouse
 
 import (
@@ -18,12 +17,16 @@ type ProfileEvent struct {
 	Value       int64
 }
 
-func (c *connect) profileEvents(ctx context.Context) ([]ProfileEvent, error) {
+func (c *connect) profileEvents(ctx context.Context, scanEvents bool) ([]ProfileEvent, error) {
 	block, err := c.readData(ctx, proto.ServerProfileEvents, false)
 	if err != nil {
 		return nil, err
 	}
 	c.debugf("[profile events] rows=%d", block.Rows())
+	if !scanEvents {
+		c.debugf("[profile events] skipping scan")
+		return nil, nil
+	}
 	var (
 		events []ProfileEvent
 		names  = block.ColumnsNames()

context.go

@@ -260,7 +260,7 @@ func queryOptionsUserLocation(ctx context.Context) *time.Location {
 }
 
 func (q *QueryOptions) onProcess() *onProcess {
-	return &onProcess{
+	onProcess := &onProcess{
 		logs: func(logs []Log) {
 			if q.events.logs != nil {
 				for _, l := range logs {
@@ -278,12 +278,16 @@ func (q *QueryOptions) onProcess() *onProcess {
 				q.events.profileInfo(p)
 			}
 		},
-		profileEvents: func(events []ProfileEvent) {
-			if q.events.profileEvents != nil {
-				q.events.profileEvents(events)
-			}
-		},
 	}
+
+	profileEventsHandler := q.events.profileEvents
+	if profileEventsHandler != nil {
+		onProcess.profileEvents = func(events []ProfileEvent) {
+			profileEventsHandler(events)
+		}
+	}
+
+	return onProcess
 }
 
 // clone returns a copy of QueryOptions where Settings and Parameters are safely mutable.

tests/issues/1685_test.go

@@ -0,0 +1,39 @@
+package issues
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/ClickHouse/clickhouse-go/v2/tests"
+	"github.com/stretchr/testify/require"
+)
+
+func benchmark1685(ctx context.Context, conn clickhouse.Conn) error {
+	for i := 0; i < 10_000; i++ {
+		err := conn.Exec(ctx,
+			"INSERT INTO test_xxxx VALUES (?, ?, [1, 2, 3, 4, 5, 6, 7, 8, 9], now())",
+			i, "Golang SQL database driver", false)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func BenchmarkIssue1685(b *testing.B) {
+	conn, err := tests.GetConnectionTCP("issues", nil, nil, nil)
+	ctx := context.Background()
+	require.NoError(b, err)
+
+	const ddl = `CREATE TABLE test_xxxx (Col1 UInt64, Col2 String, Col3 Array(UInt8), Col4 DateTime) Engine ReplacingMergeTree() ORDER BY Col1`
+	err = conn.Exec(ctx, ddl)
+	require.NoError(b, err)
+	defer func() {
+		conn.Exec(ctx, "DROP TABLE IF EXISTS test_xxxx")
+	}()
+
+	for k := 0; k < b.N; k++ {
+		require.NoError(b, benchmark1685(ctx, conn))
+	}
+}