Repository navigation
Expand file tree
/
Copy pathexecutor.go
More file actions
778 lines (738 loc) · 28.4 KB
/
Copy pathexecutor.go
File metadata and controls
778 lines (738 loc) · 28.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
package rasql
import (
"context"
"database/sql"
"fmt"
"iter"
"github.com/lestrrat-go/rasql/dialect"
"github.com/lestrrat-go/rasql/internal/bindplan"
"github.com/lestrrat-go/rasql/internal/querycompile"
"github.com/lestrrat-go/rasql/query"
"github.com/lestrrat-go/rasql/stmt"
)
type Executor interface {
Dialect() dialect.Dialect
Query(context.Context, stmt.Statement) (ResultRows, error)
Exec(context.Context, stmt.Statement) (sql.Result, error)
}
// ResultRows is what Executor.Query hands back, and it is also the interface
// a hand-written Executor must satisfy to run through Rows, All, One, Maybe,
// and Prepared. It owns exactly one query result and completes consumption
// of it exactly once, through Finish; Columns, Next, Close, Err, and
// RecordRow exist to drive that consumption, and Finish is what ends it
// regardless of how it ends.
type ResultRows interface {
// ScanSource decodes the row Next most recently advanced to, in
// database/sql's own Scan convention. An implementation that wants a
// Scan failure folded into Finish's reported error must record it itself
// when Scan returns one; nothing else does that on its behalf.
ScanSource
// Columns returns the result column names. An implementation that
// reports its own errors through Finish should fold a Columns failure
// into that accounting too, the same way it does for Scan and iteration
// failures, rather than let it vanish once the caller has already moved
// on to reading rows.
Columns() ([]string, error)
// Next advances to the next row and reports whether one is available. A
// false result means rows are exhausted or iteration failed; either way
// Next itself does not end consumption or close anything the caller can
// rely on staying open, and Err distinguishes the two causes. Once Next
// reports false it must keep reporting false, never reviving on a later
// call.
Next() bool
// Close ends consumption early, before Next has reported exhaustion. It
// must be safe to call more than once and after Next has already
// exhausted the rows, and it must leave the row source in the same state
// a Finish(nil, true) call would: an implementation typically defines one
// in terms of the other so an early Close and a caller's own Finish call
// never race to close the same result twice or report completion twice.
Close() error
// Err reports the error iteration recorded, or nil when the rows read so
// far, and any exhaustion that ended them, produced none. It must answer
// the same way before and after Finish, so a caller that already
// finished consumption can still ask what stopped it.
Err() error
// RecordRow marks the row most recently read as successfully consumed,
// separately from Scan succeeding: a caller that scans a row and then
// rejects it, for a cardinality or a decoding failure only discovered
// after Scan returns, must not call RecordRow for that row. An
// implementation that reports a row count through its own completion
// event counts only what RecordRow marked, so a caller that never calls
// it reports a count of zero rather than one derived from Next or Scan.
RecordRow()
// Finish ends consumption exactly once; every call after the first
// leaves the row source untouched and returns the same error the first
// call computed. Call it whether rows were drained to exhaustion, closed
// early, or abandoned because a decoder failed partway through: it is
// the one place that closes the underlying result if nothing has closed
// it yet, and the one place that reports this source's outcome to
// whatever observes its completion.
//
// err is the terminal cause a caller already knows about, typically a
// conversion or cardinality failure discovered after Scan already
// returned successfully. earlyClose reports whether the caller stopped
// before Next reported exhaustion; pass its true value, not always true
// or always false, since a completion observer reports whichever value
// it is given. An implementation returns err joined with whatever error
// it accumulated internally while iterating, scanning, or closing, not
// err alone, so an errors.Is or errors.As check against the err a caller
// passed in still succeeds after the join.
Finish(err error, earlyClose bool) error
}
type compilerProvider interface{ queryCompiler() *querycompile.Compiler }
type returnedColumnBinder[R any] interface {
bindReturnedColumns([]string) (RowDecoder[R], ResultSchema, error)
}
// scopeStateFrom, scopeContextFrom, and beginLogicalFrom hold the one-level
// forward decoratedExecutor uses for a capability it declares only because the
// executor it wraps declares one. compilerProvider and CodecProvider need no
// version of their own here: a decoratedExecutor always implements both
// itself, resolving its own value or this same one-level forward inline,
// so a plain type assertion on any executor in a chain already reaches the
// right answer without walking past it.
func scopeStateFrom(inner Executor) bool {
state, ok := inner.(ScopeState)
return ok && state.IsTransaction()
}
// A scope context stays forwarded rather than looked up, because Within reads
// it off the child a scope returned and takes the caller's context when the
// child reports none.
func scopeContextFrom(inner Executor) context.Context {
provider, _ := inner.(scopeContextProvider)
if provider == nil {
return nil
}
return provider.scopeContext()
}
// beginLogicalFrom starts a logical invocation on inner if it reports one,
// and hands back a no-op otherwise. A decoratedExecutor with no event
// observers of its own calls this to forward through the executor it wraps,
// then re-applies its own compiler and codecs to whatever child comes back,
// because the forwarded call has no way to know about a layer above it.
func beginLogicalFrom(executor Executor, ctx context.Context, kind EventKind) (context.Context, Executor, logicalInvocationCompletion) {
provider, ok := executor.(logicalInvocationProvider)
if !ok {
return ctx, executor, noLogicalInvocation
}
return provider.beginLogicalInvocation(ctx, kind)
}
// WithEngineProfile wraps executor so every statement it compiles is rendered
// for profile. It reports ErrInvalidEngineProfile when profile is zero and
// ErrEngineProfileMismatch when profile names a different engine than
// executor's dialect speaks.
//
// `executor` must not be nil.
func WithEngineProfile(executor Executor, profile EngineProfile) (Executor, error) {
if executor == nil {
return nil, fmt.Errorf("executor must not be nil")
}
c, err := profile.queryCompiler(executor.Dialect())
if err != nil {
return nil, err
}
return wrapProfiledChild(executor, c), nil
}
type preparedRows[R any] struct {
statement stmt.Statement
schema ResultSchema
decoder RowDecoder[R]
codecs CodecRegistry
cardinality Cardinality
emptyErr error
// parameters lists every slot a Parameter fills, in placeholder order.
// bound reports whether Prepared.Bind has supplied a value for it.
// rowsPreparedRequired refuses to run while any is false.
parameters []parameterSlot
}
type parameterSlot struct {
index int // position in the statement's arguments
id bindplan.ID
codec string
bound bool
}
type rowTerminal struct{ cause error }
type rowTerminalKey struct{}
// executorCompiler returns the compiler executor retains, reporting
// engine_profile_unavailable for one that retains none. Every caller that
// needs to know an executor is usable for compiling or matching against a
// prior compile goes through here, so the error stays the same wherever the
// check runs.
func executorCompiler(executor Executor) (*querycompile.Compiler, error) {
provider, ok := executor.(compilerProvider)
if !ok {
return nil, &PlanError{Code: "engine_profile_unavailable", Detail: "executor has no retained compiler"}
}
compiler := provider.queryCompiler()
if compiler == nil {
return nil, &PlanError{Code: "engine_profile_unavailable", Detail: "executor has no retained compiler"}
}
return compiler, nil
}
func prepareRows[R any](executor Executor, q Query[R], compiled compiledQuery) (preparedRows[R], error) {
var result preparedRows[R]
if err := q.Validate(); err != nil {
return result, err
}
composed, err := lowerQuery(q)
if err != nil {
return result, err
}
return prepareRowsLowered(executor, q, compiled, composed)
}
// prepareRowsLowered prepares q against the lowered form a caller already
// produced for it. q MUST have passed Query.Validate, and composed MUST come
// from lowerQuery called on the same q.
func prepareRowsLowered[R any](executor Executor, q Query[R], compiled compiledQuery, composed query.ResultQuery) (preparedRows[R], error) {
var result preparedRows[R]
if _, err := executorCompiler(executor); err != nil {
return result, err
}
engine, cardinality, native := q.nativeInfo()
if native {
dialect := executor.Dialect()
if dialect == nil {
return result, planError("engine_mismatch", "native.engine", "executor dialect is unavailable")
}
if dialect.Name() != engine {
return result, planError("engine_mismatch", "native.engine", "executor dialect does not match native SQL")
}
} else if q.plan.mutation == nil {
engines := querycompile.NativeEngines(composed)
for _, required := range engines {
if engine != "" && required != engine {
return result, planError("engine_mismatch", "native.engine", "composed native SQL uses multiple engines")
}
engine = required
}
if engine != "" {
dialect := executor.Dialect()
if dialect == nil || dialect.Name() != engine {
return result, planError("engine_mismatch", "native.engine", "executor dialect does not match native SQL")
}
}
}
registry, err := executorCodecs(executor)
if err != nil {
return result, err
}
columns := q.Schema().Columns()
slots := append([]bindSlot(nil), compiled.Slots...)
statementCopy, err := compiled.Copy()
if err != nil {
return result, err
}
if len(slots) != len(statementCopy.BoundArgs()) {
return result, &PlanError{Code: "bind_mismatch", Detail: "statement arguments and bind slots differ"}
}
for i, column := range columns {
if _, err := codecFor(registry, column.Codec); err != nil {
if planErr, ok := err.(*PlanError); ok {
planErr.Path = fmt.Sprintf("result.columns[%d].codec", i)
}
return result, err
}
_ = i
}
var parameters []parameterSlot
for i, slot := range slots {
if _, err := codecFor(registry, slot.Codec); err != nil {
if planErr, ok := err.(*PlanError); ok {
planErr.Path = fmt.Sprintf("binds[%d].codec", i)
}
return result, err
}
if slot.Parameter {
parameters = append(parameters, parameterSlot{index: i, id: slot.ID, codec: slot.Codec})
}
}
statement, err := encodeStatement(statementCopy, slots, registry)
if err != nil {
return result, err
}
result.statement, result.schema, result.decoder, result.codecs, result.cardinality, result.emptyErr = statement, q.Schema(), q.Projection().Decoder(), registry, cardinality, q.resultRequirement.emptyErr
result.parameters = parameters
return result, nil
}
func rowsPrepared[R any](ctx context.Context, executor Executor, prepared preparedRows[R]) (iter.Seq2[R, error], error) {
return rowsPreparedRequired(ctx, executor, prepared, Many)
}
func rowsPreparedRequired[R any](ctx context.Context, executor Executor, prepared preparedRows[R], consumer Cardinality) (iter.Seq2[R, error], error) {
for _, slot := range prepared.parameters {
if !slot.bound {
return nil, &PlanError{Code: "parameter_unbound", Path: fmt.Sprintf("binds[%d]", slot.index), Detail: "parameter has no value; call Prepared.Bind before running"}
}
}
used := false
return func(yield func(R, error) bool) {
if used {
var zero R
yield(zero, fmt.Errorf("rows sequence may only be consumed once"))
return
}
used = true
owned, err := executor.Query(ctx, prepared.statement)
var zero R
if err != nil {
yield(zero, err)
return
}
if owned == nil {
yield(zero, fmt.Errorf("executor returned nil rows"))
return
}
columns, err := owned.Columns()
if err != nil {
finished := owned.Finish(err, true)
yield(zero, finished)
return
}
expected := prepared.schema.Columns()
decoder := prepared.decoder
bound := false
if binder, ok := any(decoder).(returnedColumnBinder[R]); ok {
boundDecoder, boundSchema, bindErr := binder.bindReturnedColumns(columns)
if bindErr != nil {
finished := owned.Finish(bindErr, true)
yield(zero, finished)
return
}
decoder, expected = boundDecoder, boundSchema.Columns()
bound = true
}
if len(columns) != len(expected) {
mismatch := &PlanError{Code: "result_columns_mismatch", Path: "result.columns", Detail: "column count differs from prepared schema"}
finished := owned.Finish(mismatch, true)
yield(zero, finished)
return
}
for i, name := range columns {
if bound {
break
}
if name != expected[i].Name {
mismatch := &PlanError{Code: "result_columns_mismatch", Path: fmt.Sprintf("result.columns[%d]", i), Detail: "column name differs from prepared schema"}
finished := owned.Finish(mismatch, true)
yield(zero, finished)
return
}
}
codecs := make([]ValueCodec, len(expected))
for i, column := range expected {
codecs[i], _ = codecFor(prepared.codecs, column.Codec)
}
source := &codecScanSource{source: owned, columns: expected, codecs: codecs}
// Converted to the ScanSource interface once here rather than at each
// DecodeRow call: DecodeRow's parameter is an interface, so passing the
// concrete *codecScanSource directly would convert it to that interface
// fresh on every row.
var rowSource ScanSource = source
policy := prepared.cardinality
if consumer > policy {
policy = consumer
}
terminal, _ := ctx.Value(rowTerminalKey{}).(*rowTerminal)
// emit hands one row to the consumer. A false report means the consumer stopped early
// and consumption is already complete under whatever cause the consumer recorded, so
// the loop has nothing left to do but return.
emit := func(value R) bool {
if terminal != nil {
terminal.cause = nil
}
if yield(value, nil) {
return true
}
var cause error
if terminal != nil {
cause = terminal.cause
}
_ = owned.Finish(cause, true)
return false
}
// A single-row policy holds its row back and reads once more before yielding anything, so
// a second row raises ErrMultipleRows while the consumer has still seen no row at all.
// Many yields each row as it decodes.
singleRow := policy != Many
var held R
holding := false
for owned.Next() {
var value R
if err := decoder.DecodeRow(rowSource, &value); err != nil {
finished := owned.Finish(err, true)
yield(zero, finished)
return
}
if !singleRow {
owned.RecordRow()
if !emit(value) {
return
}
continue
}
if holding {
finished := owned.Finish(ErrMultipleRows, true)
yield(zero, finished)
return
}
owned.RecordRow()
held, holding = value, true
}
if err := owned.Err(); err != nil {
finished := owned.Finish(err, false)
yield(zero, finished)
return
}
if policy == ExactlyOne && !holding {
cause := prepared.emptyErr
if cause == nil {
cause = ErrNoRows
}
finished := owned.Finish(cause, false)
yield(zero, finished)
return
}
if holding && !emit(held) {
return
}
var terminalErr error
if terminal != nil {
terminalErr = terminal.cause
}
if err := owned.Finish(terminalErr, false); err != nil {
yield(zero, err)
}
}, nil
}
// compileAndPrepare compiles q and prepares it against executor, validating and
// lowering q once for both steps, and returns the compiler it resolved
// alongside the compiled and prepared forms. A caller that compiles one query
// and prepares a different one, as the graph loader does, calls compileQuery
// and prepareRows separately instead.
func compileAndPrepare[R any](executor Executor, q Query[R]) (compiledQuery, preparedRows[R], *querycompile.Compiler, error) {
var prepared preparedRows[R]
compiler, err := executorCompiler(executor)
if err != nil {
return compiledQuery{}, prepared, nil, err
}
if err := q.Validate(); err != nil {
return compiledQuery{}, prepared, nil, mapCompileError(err)
}
composed, err := lowerQuery(q)
if err != nil {
return compiledQuery{}, prepared, nil, err
}
compiled, err := compileQueryLowered(compiler, q, composed)
if err != nil {
return compiledQuery{}, prepared, nil, err
}
prepared, err = prepareRowsLowered(executor, q, compiled, composed)
if err != nil {
return compiledQuery{}, prepared, nil, err
}
return compiled, prepared, compiler, nil
}
// Prepared holds a query that has already been validated, lowered, rendered
// to SQL and resolved against a codec registry, so that Rows, All, One and
// Maybe below repeat none of that work across many runs of the same query.
// Prepare captures the argument values bound into q at the moment it is
// called, except for a value placed through Parameter: a Prepared carrying
// one or more unbound parameters refuses to run until Bind supplies a value
// for each, which is what lets one Prepare serve many different runs. What a
// Prepared amortizes either way is validation, lowering, rendering and codec
// lookup, not a Value-bound argument's value; call Prepare again to change
// one of those.
//
// A Prepared's SQL is rendered for the dialect the executor passed to Prepare
// spoke through its compiler, and its bind values are already encoded, and
// its rows already decode, through the codec registry that executor carried
// at that moment. Its Rows, All, One and Maybe methods take an executor
// argument again on every call, so the wrapper around it can change (entering
// a transaction scope, for instance) without re-preparing, but each checks
// that the executor it is given still carries the same compiler AND the same
// codec registry Prepare captured, and reports a prepared_executor_mismatch
// PlanError instead of silently running with a stale registry or SQL rendered
// for a different dialect. Wrapping the executor with WithCodecs after
// Prepare, in particular, means calling Prepare again rather than reusing the
// old Prepared: nothing re-reads the executor's registry per run, so the new
// codecs would otherwise never take effect.
//
// A Prepared is never mutated after Prepare returns, so it is safe to store
// and execute from multiple goroutines, and every call to Rows returns a
// fresh sequence: the "may only be consumed once" rule applies to each
// sequence Rows hands back, not to the Prepared it came from.
type Prepared[R any] struct {
compiler *querycompile.Compiler
codecs CodecRegistry
prepared preparedRows[R]
}
// Prepare validates, lowers, renders and resolves codecs for q against
// executor once, returning a Prepared[R] whose Rows, All, One and Maybe
// methods run that same work repeatedly with none of the cost. See the
// [Prepared] doc comment for what stays fixed and what a later call may vary.
//
// `executor` must not be nil.
func Prepare[R any](executor Executor, q Query[R]) (Prepared[R], error) {
_, prepared, compiler, err := compileAndPrepare(executor, q)
if err != nil {
return Prepared[R]{}, err
}
return Prepared[R]{compiler: compiler, codecs: prepared.codecs, prepared: prepared}, nil
}
// checkExecutor reports whether executor still carries the compiler AND the
// codec registry Prepare captured, so a Prepared[R] run against a different
// executor, or the same executor rewrapped with WithCodecs, fails with a
// clear error instead of sending SQL rendered for a different dialect or
// silently decoding through a registry the caller no longer intends.
func (p Prepared[R]) checkExecutor(executor Executor) error {
compiler, err := executorCompiler(executor)
if err != nil {
return err
}
if compiler != p.compiler {
return &PlanError{Code: "prepared_executor_mismatch", Detail: "executor does not match the executor Prepare was called with"}
}
codecs, err := executorCodecs(executor)
if err != nil {
return err
}
// A plain comparison is sound here because CodecRegistry is sealed: every
// registry is the one pointer type this package builds, which is always
// comparable.
if p.codecs != codecs {
return &PlanError{Code: "prepared_executor_mismatch", Detail: "executor carries a different codec registry than Prepare captured"}
}
return nil
}
// Bind returns a copy of p in which each named parameter carries its value,
// encoded through the codec registry Prepare captured. p itself is
// unchanged, so a Prepared with unbound parameters can be shared and bound
// differently by every caller.
//
// It reports invalid_parameter for a value built from a zero Parameter,
// unknown_parameter for a value naming a parameter the query does not carry,
// and duplicate_parameter for two values naming the same parameter in one
// call. A value whose Parameter.Value could not snapshot its argument is
// returned as that unsnapshotable_bind error.
func (p Prepared[R]) Bind(values ...ParameterValue) (Prepared[R], error) {
snapshots := make(map[bindplan.ID]any, len(values))
for i, value := range values {
if value.id == 0 {
return Prepared[R]{}, planError("invalid_parameter", fmt.Sprintf("params[%d]", i), "parameter must be built with NewParameter or NewParameterWithCodec")
}
if value.err != nil {
return Prepared[R]{}, value.err
}
known := false
for _, slot := range p.prepared.parameters {
if slot.id == value.id {
known = true
break
}
}
if !known {
return Prepared[R]{}, planError("unknown_parameter", fmt.Sprintf("params[%d]", i), "query has no parameter with this identity")
}
if _, duplicate := snapshots[value.id]; duplicate {
return Prepared[R]{}, planError("duplicate_parameter", fmt.Sprintf("params[%d]", i), "parameter already given a value in this call")
}
snapshots[value.id] = value.snapshot
}
args := p.prepared.statement.Args()
parameters := append([]parameterSlot(nil), p.prepared.parameters...)
encoder := bindStatementEncoder{registry: p.codecs}
for i, slot := range parameters {
snapshot, ok := snapshots[slot.id]
if !ok {
continue
}
bound := snapshot
if slot.codec != "" && snapshot != nil {
encoded, err := encoder.EncodeBind(slot.index, slot.codec, snapshot)
if err != nil {
return Prepared[R]{}, err
}
bound = encoded
}
args[slot.index] = bound
parameters[i].bound = true
}
prepared := p.prepared
prepared.statement = stmt.New(p.prepared.statement.Text(), args...)
prepared.parameters = parameters
return Prepared[R]{compiler: p.compiler, codecs: p.codecs, prepared: prepared}, nil
}
// Rows runs p against executor and returns a fresh row sequence, consumable
// once like the sequence Rows(ctx, executor, q) returns.
func (p Prepared[R]) Rows(ctx context.Context, executor Executor) (iter.Seq2[R, error], error) {
if err := p.checkExecutor(executor); err != nil {
return nil, err
}
return rowsPrepared(ctx, executor, p.prepared)
}
// All runs p against executor and collects every row into a freshly
// allocated slice. A zero-row run returns a non-nil, empty slice, not nil.
// It returns nil on error.
//
// A caller that runs the same Prepared repeatedly and wants to reuse the
// slice's storage across calls should use AppendAll instead.
func (p Prepared[R]) All(ctx context.Context, executor Executor) ([]R, error) {
values, err := p.AppendAll(ctx, make([]R, 0), executor)
if err != nil {
return nil, err
}
return values, nil
}
// AppendAll runs p against executor and appends every row to dst, returning
// the possibly-reallocated result, in the manner of the built-in append. A
// zero-row run returns dst unchanged. A failed run returns dst holding
// exactly the elements it held before the call, dropping any rows appended
// during the run, so the caller's own contents and capacity survive an
// error instead of being discarded.
//
// Passing a slice returned by an earlier AppendAll call, reset to length
// zero with dst[:0], lets repeated calls against the same Prepared reuse
// its storage instead of allocating a fresh slice each time, including
// after a call that failed.
func (p Prepared[R]) AppendAll(ctx context.Context, dst []R, executor Executor) ([]R, error) {
rows, err := p.Rows(ctx, executor)
if err != nil {
return dst, err
}
return appendRows(dst, rows)
}
// One runs p against executor and returns its single row, reporting
// ErrNoRows or ErrMultipleRows when the run does not produce exactly one.
func (p Prepared[R]) One(ctx context.Context, executor Executor) (R, error) {
var zero R
if err := p.checkExecutor(executor); err != nil {
return zero, err
}
rows, err := rowsPreparedRequired(ctx, executor, p.prepared, ExactlyOne)
if err != nil {
return zero, err
}
var result R
for value, err := range rows {
if err != nil {
return zero, err
}
result = value
}
return result, nil
}
// Maybe runs p against executor and returns its row and true when the run
// produces exactly one, or the zero value and false when it produces none.
// It reports ErrMultipleRows when the run produces more than one.
func (p Prepared[R]) Maybe(ctx context.Context, executor Executor) (R, bool, error) {
var zero R
if err := p.checkExecutor(executor); err != nil {
return zero, false, err
}
rows, err := rowsPreparedRequired(ctx, executor, p.prepared, AtMostOne)
if err != nil {
return zero, false, err
}
count := 0
var result R
for value, err := range rows {
if err != nil {
return zero, false, err
}
count++
result = value
}
return result, count == 1, nil
}
func Rows[R any](ctx context.Context, executor Executor, q Query[R]) (iter.Seq2[R, error], error) {
prepared, err := Prepare(executor, q)
if err != nil {
return nil, err
}
return rowsPrepared(ctx, executor, prepared.prepared)
}
// All runs q against executor and collects every row into a freshly
// allocated slice. A zero-row run returns a non-nil, empty slice, not nil.
// It returns nil on error.
//
// A caller that runs the same query repeatedly and wants to reuse the
// slice's storage across calls should use AppendAll instead.
func All[R any](ctx context.Context, executor Executor, q Query[R]) ([]R, error) {
values, err := AppendAll(ctx, make([]R, 0), executor, q)
if err != nil {
return nil, err
}
return values, nil
}
// AppendAll runs q against executor and appends every row to dst, returning
// the possibly-reallocated result, in the manner of the built-in append. A
// zero-row run returns dst unchanged. A failed run returns dst holding
// exactly the elements it held before the call, dropping any rows appended
// during the run, so the caller's own contents and capacity survive an
// error instead of being discarded.
//
// Passing a slice returned by an earlier AppendAll call, reset to length
// zero with dst[:0], lets repeated calls against the same query reuse its
// storage instead of allocating a fresh slice each time, including after a
// call that failed.
func AppendAll[R any](ctx context.Context, dst []R, executor Executor, q Query[R]) ([]R, error) {
rows, err := Rows(ctx, executor, q)
if err != nil {
return dst, err
}
return appendRows(dst, rows)
}
// appendRows drains rows into dst and returns the possibly-reallocated
// result. On error it returns dst truncated back to the length it had on
// entry: append may already have reallocated dst's backing array, but
// growth always copies the caller's original elements into the new array
// first, so slicing back to the starting length recovers exactly those
// elements (and, incidentally, the larger capacity). All and AppendAll
// never hand back a partial result.
func appendRows[R any](dst []R, rows iter.Seq2[R, error]) ([]R, error) {
start := len(dst)
for value, err := range rows {
if err != nil {
return dst[:start], err
}
dst = append(dst, value)
}
return dst, nil
}
func One[R any](ctx context.Context, executor Executor, q Query[R]) (R, error) {
var zero R
rows, err := rowsFor(ctx, executor, q, ExactlyOne)
if err != nil {
return zero, err
}
var result R
for value, err := range rows {
if err != nil {
return zero, err
}
result = value
}
return result, nil
}
func Maybe[R any](ctx context.Context, executor Executor, q Query[R]) (R, bool, error) {
var zero R
rows, err := rowsFor(ctx, executor, q, AtMostOne)
if err != nil {
return zero, false, err
}
count := 0
var result R
for value, err := range rows {
if err != nil {
return zero, false, err
}
count++
result = value
}
return result, count == 1, nil
}
func rowsFor[R any](ctx context.Context, executor Executor, q Query[R], consumer Cardinality) (iter.Seq2[R, error], error) {
prepared, err := Prepare(executor, q)
if err != nil {
return nil, err
}
return rowsPreparedRequired(ctx, executor, prepared.prepared, consumer)
}