forked from johnkerl/miller
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathfraction.go
296 lines (254 loc) · 9.01 KB
/
fraction.go
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
package transformers
import (
"container/list"
"errors"
"fmt"
"os"
"strings"
"github.com/johnkerl/miller/pkg/bifs"
"github.com/johnkerl/miller/pkg/cli"
"github.com/johnkerl/miller/pkg/lib"
"github.com/johnkerl/miller/pkg/mlrval"
"github.com/johnkerl/miller/pkg/types"
)
// ----------------------------------------------------------------
const verbNameFraction = "fraction"
var FractionSetup = TransformerSetup{
Verb: verbNameFraction,
UsageFunc: transformerFractionUsage,
ParseCLIFunc: transformerFractionParseCLI,
IgnoresInput: false,
}
func transformerFractionUsage(
o *os.File,
) {
argv0 := "mlr"
verb := verbNameFraction
fmt.Fprintf(o, "Usage: %s %s [options]\n", argv0, verb)
fmt.Fprintf(o, "For each record's value in specified fields, computes the ratio of that\n")
fmt.Fprintf(o, "value to the sum of values in that field over all input records.\n")
fmt.Fprintf(o, "E.g. with input records x=1 x=2 x=3 and x=4, emits output records\n")
fmt.Fprintf(o, "x=1,x_fraction=0.1 x=2,x_fraction=0.2 x=3,x_fraction=0.3 and x=4,x_fraction=0.4\n")
fmt.Fprintf(o, "\n")
fmt.Fprintf(o, "Note: this is internally a two-pass algorithm: on the first pass it retains\n")
fmt.Fprintf(o, "input records and accumulates sums; on the second pass it computes quotients\n")
fmt.Fprintf(o, "and emits output records. This means it produces no output until all input is read.\n")
fmt.Fprintf(o, "\n")
fmt.Fprintf(o, "Options:\n")
fmt.Fprintf(o, "-f {a,b,c} Field name(s) for fraction calculation\n")
fmt.Fprintf(o, "-g {d,e,f} Optional group-by-field name(s) for fraction counts\n")
fmt.Fprintf(o, "-p Produce percents [0..100], not fractions [0..1]. Output field names\n")
fmt.Fprintf(o, " end with \"_percent\" rather than \"_fraction\"\n")
fmt.Fprintf(o, "-c Produce cumulative distributions, i.e. running sums: each output\n")
fmt.Fprintf(o, " value folds in the sum of the previous for the specified group\n")
fmt.Fprintf(o, " E.g. with input records x=1 x=2 x=3 and x=4, emits output records\n")
fmt.Fprintf(o, " x=1,x_cumulative_fraction=0.1 x=2,x_cumulative_fraction=0.3\n")
fmt.Fprintf(o, " x=3,x_cumulative_fraction=0.6 and x=4,x_cumulative_fraction=1.0\n")
}
func transformerFractionParseCLI(
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++
// Parse local flags
var fractionFieldNames []string = nil
var groupByFieldNames []string = nil
doPercents := false
doCumu := false
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" {
transformerFractionUsage(os.Stdout)
os.Exit(0)
} else if opt == "-f" {
fractionFieldNames = cli.VerbGetStringArrayArgOrDie(verb, opt, args, &argi, argc)
} else if opt == "-g" {
groupByFieldNames = cli.VerbGetStringArrayArgOrDie(verb, opt, args, &argi, argc)
} else if opt == "-p" {
doPercents = true
} else if opt == "-c" {
doCumu = true
} else {
transformerFractionUsage(os.Stderr)
os.Exit(1)
}
}
if fractionFieldNames == nil {
transformerFractionUsage(os.Stderr)
os.Exit(1)
}
*pargi = argi
if !doConstruct { // All transformers must do this for main command-line parsing
return nil
}
transformer, err := NewTransformerFraction(
fractionFieldNames,
groupByFieldNames,
doPercents,
doCumu,
)
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
return transformer
}
// ----------------------------------------------------------------
type TransformerFraction struct {
fractionFieldNames []string
groupByFieldNames []string
doCumu bool
recordsAndContexts *list.List
// Two-level map: Group-by field names are the first keyset;
// fraction field names are keys into the second.
sums map[string]map[string]*mlrval.Mlrval
cumus map[string]map[string]*mlrval.Mlrval
outputFieldNameSuffix string // "_fraction" or "_percent"
multiplier *mlrval.Mlrval // 1.0 for fraction or 100.0 for percent
zero *mlrval.Mlrval
}
// ----------------------------------------------------------------
func NewTransformerFraction(
fractionFieldNames []string,
groupByFieldNames []string,
doPercents bool,
doCumu bool,
) (*TransformerFraction, error) {
recordsAndContexts := list.New()
sums := make(map[string]map[string]*mlrval.Mlrval)
cumus := make(map[string]map[string]*mlrval.Mlrval)
var multiplier *mlrval.Mlrval
var outputFieldNameSuffix string
if doPercents {
multiplier = mlrval.FromInt(100)
if doCumu {
outputFieldNameSuffix = "_cumulative_percent"
} else {
outputFieldNameSuffix = "_percent"
}
} else {
multiplier = mlrval.FromInt(1)
if doCumu {
outputFieldNameSuffix = "_cumulative_fraction"
} else {
outputFieldNameSuffix = "_fraction"
}
}
zero := mlrval.FromInt(0)
return &TransformerFraction{
fractionFieldNames: fractionFieldNames,
groupByFieldNames: groupByFieldNames,
doCumu: doCumu,
recordsAndContexts: recordsAndContexts,
sums: sums,
cumus: cumus,
outputFieldNameSuffix: outputFieldNameSuffix,
multiplier: multiplier,
zero: zero,
}, nil
}
// ----------------------------------------------------------------
func (tr *TransformerFraction) Transform(
inrecAndContext *types.RecordAndContext,
outputRecordsAndContexts *list.List, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream { // Not end of stream; pass 1
inrec := inrecAndContext.Record
// Append records into a single output list (so that this verb is order-preserving).
tr.recordsAndContexts.PushBack(inrecAndContext)
// Accumulate sums of fraction-field values grouped by group-by field names
groupingKey, hasAll := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if hasAll {
sumsForGroup := tr.sums[groupingKey]
var cumusForGroup map[string]*mlrval.Mlrval = nil
if sumsForGroup == nil {
sumsForGroup = make(map[string]*mlrval.Mlrval)
tr.sums[groupingKey] = sumsForGroup
cumusForGroup = make(map[string]*mlrval.Mlrval)
tr.cumus[groupingKey] = cumusForGroup
}
for _, fractionFieldName := range tr.fractionFieldNames {
value := inrec.Get(fractionFieldName)
if value != nil {
value.AssertNumeric() // may fatal the process
sum := sumsForGroup[fractionFieldName]
if sum == nil { // First value for group
sumsForGroup[fractionFieldName] = value.Copy()
cumusForGroup[fractionFieldName] = tr.zero
} else {
sumsForGroup[fractionFieldName] = bifs.BIF_plus_binary(sum, value)
}
}
}
}
} else { // End of stream; pass 2
// Iterate over the retained records, decorating them with fraction fields.
endOfStreamContext := inrecAndContext.Context
for {
element := tr.recordsAndContexts.Front()
if element == nil {
break
}
tr.recordsAndContexts.Remove(element)
recordAndContext := element.Value.(*types.RecordAndContext)
outrec := recordAndContext.Record
groupingKey, hasAll := outrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if hasAll {
sumsForGroup := tr.sums[groupingKey]
cumusForGroup := tr.cumus[groupingKey]
lib.InternalCodingErrorIf(sumsForGroup == nil) // should have been populated on pass 1
for _, fractionFieldName := range tr.fractionFieldNames {
value := outrec.Get(fractionFieldName)
if value != nil {
value.AssertNumeric() // may fatal the process
var numerator *mlrval.Mlrval = nil
var cumu *mlrval.Mlrval = nil
var outputValue *mlrval.Mlrval = nil
if tr.doCumu {
cumu = cumusForGroup[fractionFieldName]
numerator = bifs.BIF_plus_binary(value, cumu)
} else {
numerator = value
}
denominator := sumsForGroup[fractionFieldName]
if !mlrval.Equals(value, tr.zero) {
outputValue = bifs.BIF_divide(numerator, denominator)
outputValue = bifs.BIF_times(outputValue, tr.multiplier)
} else {
outputValue = mlrval.FromError(
errors.New(
"mlr fraction: division by zero",
),
)
}
outrec.PutCopy(
fractionFieldName+tr.outputFieldNameSuffix,
outputValue,
)
if tr.doCumu {
cumusForGroup[fractionFieldName] = bifs.BIF_plus_binary(cumu, value)
}
}
}
}
outputRecordsAndContexts.PushBack(types.NewRecordAndContext(outrec, &endOfStreamContext))
}
outputRecordsAndContexts.PushBack(inrecAndContext) // end-of-stream marker
}
}