Function Exec

flow/pkg/clickhouse/query_retry.go:65–89

(ctx context.Context, logger log.Logger,
	conn clickhouse.Conn, query string, args ...any,
)

}

func Exec(ctx context.Context, logger log.Logger,
	conn clickhouse.Conn, query string, args ...any,
) error {
	var err error
	for i := range 5 {
		err = conn.Exec(ctx, query, args...)
		if !isRetryableException(err) {
			break
		}
		logger.Info("[exec] retryable error", slog.Any("error", err), slog.Int64("retry", int64(i)))
		if i < 4 {
			time.Sleep(time.Second * time.Duration(i*5+1))
		}
	}
	if ex, ok := errors.AsType[*clickhouse.Exception](err); ok {
		isMV := strings.Contains(ex.Error(), "while pushing to view")
		if chproto.Error(ex.Code) == chproto.ErrIncorrectData {
			ex.Message = "REDACTED"
		}
		if isMV {
			return NewViewError(ex)
		}
	}
	return err
}

func Query(ctx context.Context, logger log.Logger,
	conn clickhouse.Conn, query string, args ...any,