refactor(database): simplify order_by construction and remove keep-alive logic from BulkImportStream

Author: flarco

core/dbio/database/database_clickhouse.go

@@ -331,7 +331,7 @@ func (conn *ClickhouseConn) GenerateDDL(table Table, data iop.Dataset, temporary
 	if keyCols := data.Columns.GetKeys(iop.PrimaryKey); len(keyCols) > 0 {
 		colNames := conn.GetType().QuoteNames(keyCols.Names()...)
 		primaryKey = g.F("primary key (%s)", strings.Join(colNames, ", "))
-		orderBy = g.F("(%s)", strings.Join(colNames, ", "))
+		orderBy = strings.Join(colNames, ", ")
 	}
 	ddl = g.R(ddl, "primary_key", primaryKey, "order_by", orderBy)
 
@@ -375,17 +375,13 @@ func (conn *ClickhouseConn) BulkImportStream(tableFName string, ds *iop.Datastre
 			// use pre-defined function
 			err = oldOnColumnChanged(col)
 			if err != nil {
-				return g.Error(err, "could not process ColumnChange for Postgres")
+				return g.Error(err, "could not process ColumnChange for Clickhouse")
 			}
 
 			return nil
 		}
 	}
 
-	// keep-alive interval to prevent idle connection timeout
-	keepAliveInterval := 60 * time.Second
-	lastPushTime := time.Time{}
-
 	for batch := range ds.BatchChan {
 		if batch.ColumnsChanged() || batch.IsFirst() {
 			columns, err = conn.GetColumns(tableFName, batch.Columns.Names()...)
@@ -446,42 +442,6 @@ func (conn *ClickhouseConn) BulkImportStream(tableFName string, ds *iop.Datastre
 				}
 			}
 
-			// Start a keep-alive goroutine to prevent idle connection timeout
-			// This sends a SELECT 1 query periodically to keep the connection alive
-			// during long pauses in the data stream (e.g., slow source reads)
-			stopKeepAlive := make(chan struct{})
-			keepAliveDone := make(chan struct{})
-			go func() {
-				defer close(keepAliveDone)
-				ticker := time.NewTicker(keepAliveInterval)
-				defer ticker.Stop()
-				for {
-					select {
-					case <-stopKeepAlive:
-						return
-					case <-ticker.C:
-						// Execute keep-alive query within the transaction context
-						// This keeps the connection from being closed due to idle timeout
-						ds.Context.Lock()
-						if time.Since(lastPushTime).Seconds() > keepAliveInterval.Seconds() {
-							_, keepAliveErr := conn.Tx().ExecContext(ds.Context.Ctx, "SELECT 1")
-							if keepAliveErr != nil {
-								g.Debug("keep-alive query failed: %v", keepAliveErr)
-							} else {
-								g.Trace("keep-alive query executed successfully")
-							}
-						}
-						ds.Context.Unlock()
-					}
-				}
-			}()
-
-			// Ensure we stop the keep-alive goroutine when done
-			defer func() {
-				close(stopKeepAlive)
-				<-keepAliveDone
-			}()
-
 			for row := range batch.Rows {
 				var eG g.ErrorGroup
 
@@ -553,7 +513,6 @@ func (conn *ClickhouseConn) BulkImportStream(tableFName string, ds *iop.Datastre
 				// Do insert
 				ds.Context.Lock()
 				_, err := stmt.Exec(row...)
-				lastPushTime = time.Now()
 				ds.Context.Unlock()
 				if err != nil {
 					ds.Context.CaptureErr(g.Error(err, "could not COPY into table %s", tableFName))