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
|
package transformers
import (
"container/list"
"fmt"
"os"
"strings"
"github.com/johnkerl/miller/v6/pkg/cli"
"github.com/johnkerl/miller/v6/pkg/lib"
"github.com/johnkerl/miller/v6/pkg/types"
)
// ----------------------------------------------------------------
const verbNameBootstrap = "bootstrap"
var BootstrapSetup = TransformerSetup{
Verb: verbNameBootstrap,
UsageFunc: transformerBootstrapUsage,
ParseCLIFunc: transformerBootstrapParseCLI,
IgnoresInput: false,
}
func transformerBootstrapUsage(
o *os.File,
) {
fmt.Fprintf(o, "Usage: %s %s [options]\n", "mlr", verbNameBootstrap)
fmt.Fprintf(o,
`Emits an n-sample, with replacement, of the input records.
See also %s sample and %s shuffle.
`, "mlr", "mlr")
fmt.Fprintf(o, "Options:\n")
fmt.Fprintf(o,
` -n Number of samples to output. Defaults to number of input records.
Must be non-negative.
`)
fmt.Fprintf(o, "-h|--help Show this message.\n")
}
func transformerBootstrapParseCLI(
pargi *int,
argc int,
args []string,
_ *cli.TOptions,
doConstruct bool, // false for first pass of CLI-parse, true for second pass
) IRecordTransformer {
// Skip the verb name from the current spot in the mlr command line
argi := *pargi
verb := args[argi]
argi++
nout := int64(-1)
for argi < argc /* variable increment: 1 or 2 depending on flag */ {
opt := args[argi]
if !strings.HasPrefix(opt, "-") {
break // No more flag options to process
}
if args[argi] == "--" {
break // All transformers must do this so main-flags can follow verb-flags
}
argi++
if opt == "-h" || opt == "--help" {
transformerBootstrapUsage(os.Stdout)
os.Exit(0)
} else if opt == "-n" {
nout = cli.VerbGetIntArgOrDie(verb, opt, args, &argi, argc)
} else {
transformerBootstrapUsage(os.Stderr)
os.Exit(1)
}
}
*pargi = argi
if !doConstruct { // All transformers must do this for main command-line parsing
return nil
}
transformer, err := NewTransformerBootstrap(nout)
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
return transformer
}
// ----------------------------------------------------------------
type TransformerBootstrap struct {
recordsAndContexts *list.List
nout int64
}
func NewTransformerBootstrap(nout int64) (*TransformerBootstrap, error) {
tr := &TransformerBootstrap{
recordsAndContexts: list.New(),
nout: nout,
}
return tr, nil
}
// ----------------------------------------------------------------
func (tr *TransformerBootstrap) Transform(
inrecAndContext *types.RecordAndContext,
outputRecordsAndContexts *list.List, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
// Not end of input stream: retain the record, and emit nothing until end of stream.
if !inrecAndContext.EndOfStream {
tr.recordsAndContexts.PushBack(inrecAndContext)
return
}
// Else end of record stream
// Given nin input records, we produce nout output records, but
// sampling with replacement.
//
// About memory management:
//
// Normally in Miller transformers we pass through pointers to records.
// Here, though, since we do sampling with replacement, a record could
// be emitted twice or more. To avoid producing multiple records in the
// output stream pointing to the same memory, we would have to copy the
// second one. In the original C (single-threaded) version of this
// code, that was the case.
//
// However, in Go, there is concurrent processing. It would be
// possible for us to emit a pointer to a particular record without
// copying, then when emitting that same record a second time, copy it.
// But due to concurrency, the pointed-to record could have already
// been mutated downstream. We wouldn't be copying our input as we
// received it -- we'd be copying something potentially modified.
//
// For that reason, this transformer must copy all output.
// TODO: Go list Len() maxes at 2^31. We should track this ourselves in an int.
nin := int64(tr.recordsAndContexts.Len())
nout := tr.nout
if nout == -1 {
nout = nin
}
if nout == 0 {
// Emit the stream-terminating null record
outputRecordsAndContexts.PushBack(inrecAndContext)
return
}
// Make an array of pointers into the input list.
recordArray := make([]*types.RecordAndContext, nin)
for i := int64(0); i < nin; i++ {
head := tr.recordsAndContexts.Front()
if head == nil {
break
}
recordArray[i] = head.Value.(*types.RecordAndContext)
tr.recordsAndContexts.Remove(head)
}
// Do the sample-with-replacment, reading from random indices in the input
// array and emitting output.
for i := int64(0); i < nout; i++ {
index := lib.RandRange(0, nin)
recordAndContext := recordArray[index]
// Already emitted once; copy
outputRecordsAndContexts.PushBack(recordAndContext.Copy())
}
// Emit the stream-terminating null record
outputRecordsAndContexts.PushBack(inrecAndContext)
}
|
