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
|
package types
import (
"bytes"
"container/list"
"strconv"
"github.com/johnkerl/miller/v6/pkg/mlrval"
)
// Since Go is concurrent, the context struct (AWK-like variables such as
// FILENAME, NF, NR, FNR, etc.) needs to be duplicated and passed through the
// channels along with each record.
//
// Strings to be printed from put/filter DSL print/dump/etc statements are
// passed along to the output channel via this OutputString rather than
// fmt.Println directly in the put/filter handlers since we want all print
// statements and record-output to be in the same goroutine, for deterministic
// output ordering.
type RecordAndContext struct {
Record *mlrval.Mlrmap
Context Context
OutputString string
EndOfStream bool
}
func NewRecordAndContext(
record *mlrval.Mlrmap,
context *Context,
) *RecordAndContext {
return &RecordAndContext{
Record: record,
// Since Go is concurrent, the context struct needs to be duplicated and
// passed through the channels along with each record. Here is where
// the copy happens, via the '*' in *context.
Context: *context,
OutputString: "",
EndOfStream: false,
}
}
// For the record-readers to update their initial context as each new record is read.
func (rac *RecordAndContext) Copy() *RecordAndContext {
if rac == nil {
return nil
}
recordCopy := rac.Record.Copy()
contextCopy := rac.Context
return &RecordAndContext{
Record: recordCopy,
Context: contextCopy,
OutputString: "",
EndOfStream: false,
}
}
// For print/dump/etc to insert strings sequenced into the record-output
// stream. This avoids race conditions between different goroutines printing
// to stdout: we have a single designated goroutine printing to stdout. This
// makes output more predictable and intuitive for users; it also makes our
// regression tests run reliably the same each time.
func NewOutputString(
outputString string,
context *Context,
) *RecordAndContext {
return &RecordAndContext{
Record: nil,
Context: *context,
OutputString: outputString,
EndOfStream: false,
}
}
// For the record-readers to update their initial context as each new record is read.
func NewEndOfStreamMarker(context *Context) *RecordAndContext {
return &RecordAndContext{
Record: nil,
Context: *context,
OutputString: "",
EndOfStream: true,
}
}
// TODO: comment
// For the record-readers to update their initial context as each new record is read.
func NewEndOfStreamMarkerList(context *Context) *list.List {
ell := list.New()
ell.PushBack(NewEndOfStreamMarker(context))
return ell
}
// ----------------------------------------------------------------
type Context struct {
FILENAME string
FILENUM int64
// This is computed dynammically from the current record's field-count
// NF int
NR int64
FNR int64
// XXX 1513
JSONHadBrackets bool
}
// TODO: comment: Remember command-line values to pass along to CST evaluators.
// The options struct-pointer can be nil when invoked by non-DSL verbs such as
// join or seqgen.
func NewContext() *Context {
context := &Context{
FILENAME: "(stdin)",
FILENUM: 0,
NR: 0,
FNR: 0,
}
return context
}
// TODO: comment: Remember command-line values to pass along to CST evaluators.
// The options struct-pointer can be nil when invoked by non-DSL verbs such as
// join or seqgen.
func NewNilContext() *Context { // TODO: rename
context := &Context{
FILENAME: "(stdin)",
FILENUM: 0,
NR: 0,
FNR: 0,
}
return context
}
// For the record-readers to update their initial context as each new file is opened.
func (context *Context) UpdateForStartOfFile(filename string) {
context.FILENAME = filename
context.FILENUM++
context.FNR = 0
}
// For the record-readers to update their initial context as each new record is read.
func (context *Context) UpdateForInputRecord() {
context.NR++
context.FNR++
}
func (context *Context) Copy() *Context {
other := *context
return &other
}
func (context *Context) GetStatusString() string {
var buffer bytes.Buffer // stdio is non-buffered in Go, so buffer for speed increase
buffer.WriteString("FILENAME=\"")
buffer.WriteString(context.FILENAME)
buffer.WriteString("\",FILENUM=")
buffer.WriteString(strconv.FormatInt(context.FILENUM, 10))
buffer.WriteString(",NR=")
buffer.WriteString(strconv.FormatInt(context.NR, 10))
buffer.WriteString(",FNR=")
buffer.WriteString(strconv.FormatInt(context.FNR, 10))
return buffer.String()
}
|
