mirror of
https://github.com/hanzo-ds/go.git
synced 2026-10-30 19:45:01 +00:00
zeekay
22b3f5ce2b
Feature-complete v2.44.0 (lib/churl, lib/col, BFloat16, QBit, scanTypeDuration, Options.Logger *slog.Logger) — fixes the [email protected].0 downgrade RED caught. Package datastore, module github.com/hanzo-ds/go, depends only on hanzo-ds/native. Wire strings preserved VERBATIM (X-ClickHouse-* headers, sql.Register("clickhouse"), clickhouse:// DSN, ClientName="clickhouse-go") — consistent with native.
80 lines
1.8 KiB
Go
80 lines
1.8 KiB
Go
package datastore
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
|
|
"github.com/hanzo-ds/go/lib/proto"
|
|
)
|
|
|
|
func (c *connect) query(ctx context.Context, release nativeTransportRelease, query string, args ...any) (*rows, error) {
|
|
var (
|
|
options = queryOptions(ctx)
|
|
onProcess = options.onProcess()
|
|
queryParamsProtocolSupport = c.revision >= proto.DBMS_MIN_PROTOCOL_VERSION_WITH_PARAMETERS
|
|
body, err = bindQueryOrAppendParameters(queryParamsProtocolSupport, &options, query, c.server.Timezone, args...)
|
|
)
|
|
|
|
if err != nil {
|
|
c.logger.Error("failed to bind query parameters", slog.Any("error", err))
|
|
release(c, err)
|
|
return nil, err
|
|
}
|
|
|
|
if err = c.sendQuery(body, &options); err != nil {
|
|
release(c, err)
|
|
return nil, err
|
|
}
|
|
|
|
init, err := c.firstBlock(ctx, onProcess)
|
|
|
|
if err != nil {
|
|
c.logger.Error("failed to get first block", slog.Any("error", err))
|
|
release(c, err)
|
|
return nil, err
|
|
}
|
|
bufferSize := c.blockBufferSize
|
|
if options.blockBufferSize > 0 {
|
|
// allow block buffer sze to be overridden per query
|
|
bufferSize = options.blockBufferSize
|
|
}
|
|
var (
|
|
errors = make(chan error, 1)
|
|
stream = make(chan *proto.Block, bufferSize)
|
|
)
|
|
|
|
go func() {
|
|
onProcess.data = func(b *proto.Block) {
|
|
stream <- b
|
|
}
|
|
err := c.process(ctx, onProcess)
|
|
if err != nil {
|
|
c.logger.Error("query processing failed", slog.Any("error", err))
|
|
errors <- err
|
|
}
|
|
close(stream)
|
|
close(errors)
|
|
release(c, err)
|
|
}()
|
|
|
|
return &rows{
|
|
block: init,
|
|
stream: stream,
|
|
errors: errors,
|
|
columns: init.ColumnsNames(),
|
|
structMap: c.structMap,
|
|
}, nil
|
|
}
|
|
|
|
func (c *connect) queryRow(ctx context.Context, release nativeTransportRelease, query string, args ...any) *row {
|
|
rows, err := c.query(ctx, release, query, args...)
|
|
if err != nil {
|
|
return &row{
|
|
err: err,
|
|
}
|
|
}
|
|
return &row{
|
|
rows: rows,
|
|
}
|
|
}
|