Mirrors/miller
1
0
Fork 0
mirror of https://github.com/johnkerl/miller.git synced 2026-07-30 11:10:11 +00:00
Code Issues Projects Releases Wiki Activity

end-of-stream nilrec -> bool refactor

Browse source
This commit is contained in:
John Kerl 2021-01-10 12:11:02 -05:00
parent 8c3151c691
commit 3dd0b02419
50 changed files with 279 additions and 226 deletions
Download patch file Download diff file Expand all files Collapse all files

8
go/README.md
View file

@ -135,16 +135,16 @@ Through out the code, records are passed by reference (as are most things, for
that matter, to reduce unnecessary data copies). In particular, records can be
nil through the reader/transformer/writer sequence.
* Record-readers produce a nil record-pointer to signify end of input stream.
* Record-readers produce an end-of-stream marker (within the `RecordAndContext` struct) to signify end of input stream.
* Each transformer takes a record-pointer as input and produces a sequence of zero or more record-pointers.
* Many transformers, such as `cat`, `cut`, `rename`, etc. produce one output record per input record.
* The `filter` transformer produces one or zero output records per input record depending on whether the record passed the filter.
* The `nothing` transformer produces zero output records.
* The `sort` and `tac` transformers are *non-streaming* -- they produce zero output records per input record, and instead retain each input record in a list. Then, when the nil-record end-of-stream marker is received, they sort/reverse the records and emit them, then they emit the nil-record end-of-stream marker.
* The `sort` and `tac` transformers are *non-streaming* -- they produce zero output records per input record, and instead retain each input record in a list. Then, when the end-of-stream marker is received, they sort/reverse the records and emit them, then they emit the end-of-stream marker.
* Many transformers such as `stats1` and `count` also retain input records, then produce output once there is no more input to them.
* A null record-pointer at end of stream is passed to record-writers so that they may produce final output.
* An end-of-stream marker is passed to record-writers so that they may produce final output.
* Most writers produce their output one record at a time.
* The pretty-print writer produces no output until end of stream, since it needs to compute the max width down each column.
* The pretty-print writer produces no output until end of stream (or schema change), since it needs to compute the max width down each column.
## Memory management

7
go/src/miller/input/record_reader_csv.go
View file

@ -54,10 +54,7 @@ func (this *RecordReaderCSV) Read(
}
}
}
inputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&context,
)
inputChannel <- types.NewEndOfStreamMarker(&context)
}
// ----------------------------------------------------------------
@ -161,7 +158,7 @@ func (this *RecordReaderCSV) processHandle(
}
}
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(
record,

9
go/src/miller/input/record_reader_csvlite.go
View file

@ -116,10 +116,7 @@ func (this *RecordReaderCSVLite) Read(
}
}
}
inputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&context,
)
inputChannel <- types.NewEndOfStreamMarker(&context)
}
// ----------------------------------------------------------------
@ -203,7 +200,7 @@ func (this *RecordReaderCSVLite) processHandleExplicitCSVHeader(
}
}
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(
record,
context,
@ -296,7 +293,7 @@ func (this *RecordReaderCSVLite) processHandleImplicitCSVHeader(
}
}
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(
record,
context,

7
go/src/miller/input/record_reader_dkvp.go
View file

@ -47,10 +47,7 @@ func (this *RecordReaderDKVP) Read(
}
}
}
inputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&context,
)
inputChannel <- types.NewEndOfStreamMarker(&context)
}
func (this *RecordReaderDKVP) processHandle(
@ -75,7 +72,7 @@ func (this *RecordReaderDKVP) processHandle(
// This is how to do a chomp:
line = strings.TrimRight(line, "\n")
record := recordFromDKVPLine(&line, &this.ifs, &this.ips)
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(
record,
context,

9
go/src/miller/input/record_reader_json.go
View file

@ -40,10 +40,7 @@ func (this *RecordReaderJSON) Read(
}
}
}
inputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&context,
)
inputChannel <- types.NewEndOfStreamMarker(&context)
}
func (this *RecordReaderJSON) processHandle(
@ -78,7 +75,7 @@ func (this *RecordReaderJSON) processHandle(
errorChannel <- errors.New("Internal coding error detected in JSON record-reader")
return
}
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(
record,
context,
@ -105,7 +102,7 @@ func (this *RecordReaderJSON) processHandle(
errorChannel <- errors.New("Internal coding error detected in JSON record-reader")
return
}
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(
record,
context,

7
go/src/miller/input/record_reader_nidx.go
View file

@ -47,10 +47,7 @@ func (this *RecordReaderNIDX) Read(
}
}
}
inputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&context,
)
inputChannel <- types.NewEndOfStreamMarker(&context)
}
func (this *RecordReaderNIDX) processHandle(
@ -77,7 +74,7 @@ func (this *RecordReaderNIDX) processHandle(
line = strings.TrimRight(line, "\n")
record := recordFromNIDXLine(&line, &this.ifs)
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(
record,
context,

9
go/src/miller/input/record_reader_xtab.go
View file

@ -62,10 +62,7 @@ func (this *RecordReaderXTAB) Read(
}
}
}
inputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&context,
)
inputChannel <- types.NewEndOfStreamMarker(&context)
}
func (this *RecordReaderXTAB) processHandle(
@ -94,7 +91,7 @@ func (this *RecordReaderXTAB) processHandle(
errorChannel <- err
return
}
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(record, context)
linesForRecord = list.New()
}
@ -111,7 +108,7 @@ func (this *RecordReaderXTAB) processHandle(
errorChannel <- err
return
}
context.UpdateForInputRecord(record)
context.UpdateForInputRecord()
inputChannel <- types.NewRecordAndContext(record, context)
linesForRecord = list.New()
}

32
go/src/miller/output/channel_writer.go
View file

@ -1,6 +1,7 @@
package output
import (
"fmt"
"os"
"miller/types"
@ -14,9 +15,34 @@ func ChannelWriter(
) {
for {
recordAndContext := <-outputChannel
record := recordAndContext.Record
recordWriter.Write(record)
if record == nil { // end of record stream
// Three things can come through:
// * End-of-stream marker
// * Non-nil records to be printed
// * Strings to be printed from put/filter DSL print/dump/etc
// statements. They are handled here 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.
if !recordAndContext.EndOfStream {
record := recordAndContext.Record
if record != nil {
recordWriter.Write(record)
}
outputString := recordAndContext.OutputString
if outputString != "" {
fmt.Print(outputString)
}
} else {
// Let the record-writers drain their output, if they have any
// queued up. For example, PPRINT needs to see all same-schema
// records before printing any, since it needs to compute max width
// down columns.
recordWriter.Write(nil)
done <- true
break
}

4
go/src/miller/transformers/altkv.go
View file

@ -85,8 +85,8 @@ func (this *TransformerAltkv) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
newrec := types.NewMlrmapAsRecord()
outputFieldNumber := 1

109
go/src/miller/transformers/bootstrap.go
View file

@ -107,66 +107,67 @@ func (this *TransformerBootstrap) Transform(
outputChannel chan<- *types.RecordAndContext,
) {
// Not end of input stream: retain the record, and emit nothing until end of stream.
if inrecAndContext.Record != nil {
if !inrecAndContext.EndOfStream {
this.recordsAndContexts.PushBack(inrecAndContext)
return
}
} else { // end of record stream
// 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 saem 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.
// 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 saem 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 int64.
nin := this.recordsAndContexts.Len()
nout := this.nout
if nout == -1 {
nout = nin
}
if nout == 0 {
// Emit the stream-terminating null record
outputChannel <- inrecAndContext
return
}
// Make an array of pointers into the input list.
recordArray := make([]*types.RecordAndContext, nin)
for i := 0; i < nin; i++ {
head := this.recordsAndContexts.Front()
if head == nil {
break
}
recordArray[i] = head.Value.(*types.RecordAndContext)
this.recordsAndContexts.Remove(head)
}
// Do the sample-with-replacment, reading from random indices in the input
// array and emitting output.
for i := 0; i < nout; i++ {
index := lib.RandRange(0, nin)
recordAndContext := recordArray[index]
// Already emitted once; copy
outputChannel <- recordAndContext.Copy()
}
// TODO: Go list Len() maxes at 2^31. We should track this ourselves in an int64.
nin := this.recordsAndContexts.Len()
nout := this.nout
if nout == -1 {
nout = nin
}
if nout == 0 {
// Emit the stream-terminating null record
outputChannel <- inrecAndContext
return
}
// Make an array of pointers into the input list.
recordArray := make([]*types.RecordAndContext, nin)
for i := 0; i < nin; i++ {
head := this.recordsAndContexts.Front()
if head == nil {
break
}
recordArray[i] = head.Value.(*types.RecordAndContext)
this.recordsAndContexts.Remove(head)
}
// Do the sample-with-replacment, reading from random indices in the input
// array and emitting output.
for i := 0; i < nout; i++ {
index := lib.RandRange(0, nin)
recordAndContext := recordArray[index]
// Already emitted once; copy
outputChannel <- recordAndContext.Copy()
}
// Emit the stream-terminating null record
outputChannel <- inrecAndContext
}

8
go/src/miller/transformers/cat.go
View file

@ -165,8 +165,8 @@ func (this *TransformerCat) countersUngrouped(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
this.counter++
key := this.counterFieldName
value := types.MlrvalFromInt64(this.counter)
@ -180,8 +180,8 @@ func (this *TransformerCat) countersGrouped(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
var counter int64 = 0

2
go/src/miller/transformers/check.go
View file

@ -84,7 +84,7 @@ func (this *TransformerCheck) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
if inrecAndContext.Record == nil { // end of stream
if inrecAndContext.EndOfStream {
outputChannel <- inrecAndContext
}
}

4
go/src/miller/transformers/count-similar.go
View file

@ -126,8 +126,8 @@ func (this *TransformerCountSimilar) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
if !ok { // This particular record doesn't have the specified fields; ignore

7
go/src/miller/transformers/count.go
View file

@ -161,8 +161,7 @@ func (this *TransformerCount) countUngrouped(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
this.ungroupedCount++
} else {
newrec := types.NewMlrmapAsRecord()
@ -180,8 +179,8 @@ func (this *TransformerCount) countGrouped(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, selectedValues, ok := inrec.GetSelectedValuesAndJoined(
this.groupByFieldNameList,

12
go/src/miller/transformers/cut.go
View file

@ -193,8 +193,8 @@ func (this *TransformerCut) includeWithInputOrder(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
outrec := types.NewMlrmap()
for pe := inrec.Head; pe != nil; pe = pe.Next {
fieldName := *pe.Key
@ -216,8 +216,8 @@ func (this *TransformerCut) includeWithArgOrder(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
outrec := types.NewMlrmap()
for _, fieldName := range this.fieldNameList {
value := inrec.Get(&fieldName)
@ -238,8 +238,8 @@ func (this *TransformerCut) exclude(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for _, fieldName := range this.fieldNameList {
if inrec.Has(&fieldName) {
inrec.Remove(&fieldName)

4
go/src/miller/transformers/decimate.go
View file

@ -143,8 +143,8 @@ func (this *TransformerDecimate) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
if !ok {

4
go/src/miller/transformers/fill-down.go
View file

@ -125,8 +125,8 @@ func (this *TransformerFillDown) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for _, fillDownFieldName := range this.fillDownFieldNames {
present := false

8
go/src/miller/transformers/flatten.go
View file

@ -137,8 +137,8 @@ func (this *TransformerFlatten) flattenAll(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
oFlatSep := this.oFlatSep
if oFlatSep == "" {
oFlatSep = inrecAndContext.Context.OFLATSEP
@ -155,8 +155,8 @@ func (this *TransformerFlatten) flattenSome(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
oFlatSep := this.oFlatSep
if oFlatSep == "" {
oFlatSep = inrecAndContext.Context.OFLATSEP

8
go/src/miller/transformers/gap.go
View file

@ -140,9 +140,7 @@ func (this *TransformerGap) mapUnkeyed(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
if this.recordCount > 0 && this.recordCount%this.gapCount == 0 {
newrec := types.NewMlrmapAsRecord()
outputChannel <- types.NewRecordAndContext(newrec, &inrecAndContext.Context)
@ -160,8 +158,8 @@ func (this *TransformerGap) mapKeyed(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
if !ok {

4
go/src/miller/transformers/grep.go
View file

@ -144,8 +144,8 @@ func (this *TransformerGrep) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
inrecAsString := inrec.ToDKVPString()
matches := this.regexp.Match([]byte(inrecAsString))
if this.invert {

4
go/src/miller/transformers/group-by.go
View file

@ -114,8 +114,8 @@ func (this *TransformerGroupBy) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
if !ok {

4
go/src/miller/transformers/group-like.go
View file

@ -94,8 +94,8 @@ func (this *TransformerGroupLike) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey := inrec.GetKeysJoined()

7
go/src/miller/transformers/head.go
View file

@ -146,8 +146,7 @@ func (this *TransformerHead) mapUnkeyed(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
this.unkeyedRecordCount++
if this.unkeyedRecordCount <= this.headCount {
outputChannel <- inrecAndContext
@ -161,8 +160,8 @@ func (this *TransformerHead) mapKeyed(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
if !ok {

2
go/src/miller/transformers/join-bucket-keeper.go
View file

@ -575,7 +575,7 @@ func (this *tJoinBucketKeeper) readRecord() *types.RecordAndContext {
fmt.Fprintln(os.Stderr, os.Args[0], ": ", err)
os.Exit(1)
case leftrecAndContext := <-this.inputChannel:
if leftrecAndContext.Record == nil { // end-of-stream marker
if leftrecAndContext.EndOfStream { // end-of-stream marker
this.recordReaderDone = true
return nil
} else {

13
go/src/miller/transformers/join.go
View file

@ -364,8 +364,8 @@ func (this *TransformerJoin) transformHalfStreaming(
this.ingested = true
}
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, hasAllJoinKeys := inrec.GetSelectedValuesJoined(
this.opts.rightJoinFieldNames,
)
@ -406,9 +406,8 @@ func (this *TransformerJoin) transformDoublyStreaming(
) {
keeper := this.joinBucketKeeper // keystroke-saver
rightRec := rightRecAndContext.Record
if rightRec != nil { // not end of record stream
if !rightRecAndContext.EndOfStream {
rightRec := rightRecAndContext.Record
isPaired := false
rightFieldValues, hasAllJoinKeys := rightRec.ReferenceSelectedValues(
@ -493,11 +492,11 @@ func (this *TransformerJoin) ingestLeftFile() {
os.Exit(1)
case leftrecAndContext := <-inputChannel:
leftrec := leftrecAndContext.Record
if leftrec == nil { // end-of-stream marker
if leftrecAndContext.EndOfStream {
done = true
break // breaks the switch, not the for, in Golang
}
leftrec := leftrecAndContext.Record
groupingKey, leftFieldValues, ok := leftrec.GetSelectedValuesAndJoined(
this.opts.leftJoinFieldNames,

8
go/src/miller/transformers/json-parse.go
View file

@ -127,8 +127,8 @@ func (this *TransformerJSONParse) jsonParseAll(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
pe.JSONParseInPlace()
}
@ -143,8 +143,8 @@ func (this *TransformerJSONParse) jsonParseSome(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if this.fieldNameSet[*pe.Key] {
pe.JSONParseInPlace()

8
go/src/miller/transformers/json-stringify.go
View file

@ -148,8 +148,8 @@ func (this *TransformerJSONStringify) jsonStringifyAll(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
pe.JSONStringifyInPlace(this.jsonFormatting)
}
@ -164,8 +164,8 @@ func (this *TransformerJSONStringify) jsonStringifySome(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if this.fieldNameSet[*pe.Key] {
pe.JSONStringifyInPlace(this.jsonFormatting)

6
go/src/miller/transformers/label.go
View file

@ -127,9 +127,9 @@ func (this *TransformerLabel) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
inrec.Label(this.newNames)
}
outputChannel <- inrecAndContext // end-of-stream marker
outputChannel <- inrecAndContext // including end-of-stream marker
}

2
go/src/miller/transformers/nothing.go
View file

@ -83,7 +83,7 @@ func (this *TransformerNothing) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
if inrecAndContext.Record == nil { // end of stream
if inrecAndContext.EndOfStream {
outputChannel <- inrecAndContext
}
}

8
go/src/miller/transformers/put_or_filter.go
View file

@ -380,7 +380,7 @@ func (this *TransformerPut) Transform(
inrec := inrecAndContext.Record
context := inrecAndContext.Context
if inrec != nil {
if !inrecAndContext.EndOfStream {
// Execute the begin { ... } before the first input record
this.callCount++
@ -433,10 +433,6 @@ func (this *TransformerPut) Transform(
os.Exit(1)
}
outputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&context,
)
outputChannel <- types.NewEndOfStreamMarker(&context)
}
}

4
go/src/miller/transformers/regularize.go
View file

@ -92,8 +92,8 @@ func (this *TransformerRegularize) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
currentFieldNames := inrec.GetKeys()
currentSortedFieldNames := lib.SortedStrings(currentFieldNames)
currentSortedFieldNamesJoined := strings.Join(currentSortedFieldNames, ",")

4
go/src/miller/transformers/remove-empty-columns.go
View file

@ -90,8 +90,8 @@ func (this *TransformerRemoveEmptyColumns) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
this.recordsAndContexts.PushBack(inrecAndContext)
for pe := inrec.Head; pe != nil; pe = pe.Next {

6
go/src/miller/transformers/rename.go
View file

@ -119,8 +119,8 @@ func (this *TransformerRename) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if this.oldToNewNames.Has(*pe.Key) {
@ -130,5 +130,5 @@ func (this *TransformerRename) Transform(
}
}
outputChannel <- inrecAndContext // end-of-stream marker
outputChannel <- inrecAndContext // including end-of-stream marker
}

8
go/src/miller/transformers/reorder.go
View file

@ -144,8 +144,8 @@ func (this *TransformerReorder) reorderToStart(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for _, fieldName := range this.fieldNameList {
inrec.MoveToHead(&fieldName)
}
@ -161,8 +161,8 @@ func (this *TransformerReorder) reorderToEnd(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for _, fieldName := range this.fieldNameList {
inrec.MoveToTail(&fieldName)
}

4
go/src/miller/transformers/sample.go
View file

@ -126,9 +126,9 @@ func (this *TransformerSample) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
// Not end of input stream: retain the record, and emit nothing until end of stream.
if inrec != nil {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
if ok {
sampleBucket := this.bucketsByGroup.Get(groupingKey)

6
go/src/miller/transformers/sec2gmt.go
View file

@ -141,8 +141,8 @@ func (this *TransformerSec2GMT) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // Not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for _, fieldName := range this.fieldNameList {
value := inrec.Get(&fieldName)
if value != nil {
@ -153,7 +153,7 @@ func (this *TransformerSec2GMT) Transform(
}
}
}
outputChannel <- inrecAndContext // end-of-stream marker
outputChannel <- inrecAndContext
} else { // End of record stream
outputChannel <- inrecAndContext // end-of-stream marker

5
go/src/miller/transformers/seqgen.go
View file

@ -203,7 +203,7 @@ func (this *TransformerSeqgen) Transform(
outrec := types.NewMlrmapAsRecord()
outrec.PutCopy(&this.fieldName, &counter)
context.UpdateForInputRecord(outrec)
context.UpdateForInputRecord()
outrecAndContext := types.NewRecordAndContext(outrec, context)
outputChannel <- outrecAndContext
@ -211,6 +211,5 @@ func (this *TransformerSeqgen) Transform(
counter = types.MlrvalBinaryPlus(&counter, &this.step)
}
outrecAndContext := types.NewRecordAndContext(nil, context)
outputChannel <- outrecAndContext
outputChannel <- types.NewEndOfStreamMarker(context)
}

2
go/src/miller/transformers/shuffle.go
View file

@ -96,7 +96,7 @@ func (this *TransformerShuffle) Transform(
outputChannel chan<- *types.RecordAndContext,
) {
// Not end of input stream: retain the record, and emit nothing until end of stream.
if inrecAndContext.Record != nil {
if !inrecAndContext.EndOfStream {
this.recordsAndContexts.PushBack(inrecAndContext)
} else { // end of record stream

4
go/src/miller/transformers/skip-trivial-records.go
View file

@ -85,8 +85,8 @@ func (this *TransformerSkipTrivialRecords) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
hasAny := false
for pe := inrec.Head; pe != nil; pe = pe.Next {
if pe.Value.String() != "" {

6
go/src/miller/transformers/sort-within-records.go
View file

@ -91,9 +91,9 @@ func (this *TransformerSortWithinRecords) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
inrec.SortByKey()
}
outputChannel <- inrecAndContext // end-of-stream marker
outputChannel <- inrecAndContext // including end-of-stream marker
}

4
go/src/miller/transformers/sort.go
View file

@ -250,8 +250,8 @@ func (this *TransformerSort) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // Not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, selectedValues, ok := inrec.GetSelectedValuesAndJoined(
this.groupByFieldNameList,

2
go/src/miller/transformers/stats1.go
View file

@ -276,7 +276,7 @@ func (this *TransformerStats1) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
if inrecAndContext.Record != nil {
if !inrecAndContext.EndOfStream {
this.handleInputRecord(inrecAndContext, outputChannel)
} else {
this.handleEndOfRecordStream(inrecAndContext, outputChannel)

6
go/src/miller/transformers/step.go
View file

@ -249,13 +249,13 @@ func (this *TransformerStep) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec == nil { // end of record stream
if inrecAndContext.EndOfStream {
outputChannel <- inrecAndContext
return
}
inrec := inrecAndContext.Record
// Group-by field names are ["a", "b"]
// Input data {"a": "s", "b": "t", "x": 3.4, "y": 5.6}
// Grouping key is "s,t"

7
go/src/miller/transformers/tac.go
View file

@ -85,16 +85,13 @@ func (this *TransformerTac) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
if inrecAndContext.Record != nil {
if !inrecAndContext.EndOfStream {
this.recordsAndContexts.PushFront(inrecAndContext)
} else {
// end of stream
for e := this.recordsAndContexts.Front(); e != nil; e = e.Next() {
outputChannel <- e.Value.(*types.RecordAndContext)
}
outputChannel <- types.NewRecordAndContext(
nil, // signals end of input record stream
&inrecAndContext.Context,
)
outputChannel <- types.NewEndOfStreamMarker(&inrecAndContext.Context)
}
}

4
go/src/miller/transformers/tail.go
View file

@ -126,8 +126,8 @@ func (this *TransformerTail) Transform(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(this.groupByFieldNameList)
if !ok {

8
go/src/miller/transformers/unflatten.go
View file

@ -136,8 +136,8 @@ func (this *TransformerUnflatten) unflattenAll(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
iFlatSep := this.iFlatSep
if iFlatSep == "" {
iFlatSep = inrecAndContext.Context.IFLATSEP
@ -154,8 +154,8 @@ func (this *TransformerUnflatten) unflattenSome(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
iFlatSep := this.iFlatSep
if iFlatSep == "" {
iFlatSep = inrecAndContext.Context.IFLATSEP

10
go/src/miller/transformers/unsparsify.go
View file

@ -143,8 +143,8 @@ func (this *TransformerUnsparsify) mapNonStreaming(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
key := *pe.Key
if !this.fieldNamesSeen.Has(key) {
@ -179,8 +179,8 @@ func (this *TransformerUnsparsify) mapStreaming(
inrecAndContext *types.RecordAndContext,
outputChannel chan<- *types.RecordAndContext,
) {
inrec := inrecAndContext.Record
if inrec != nil { // not end of record stream
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := this.fieldNamesSeen.Head; pe != nil; pe = pe.Next {
if !inrec.Has(&pe.Key) {
@ -188,7 +188,7 @@ func (this *TransformerUnsparsify) mapStreaming(
}
}
outputChannel <- inrecAndContext // end-of-stream marker
outputChannel <- inrecAndContext
} else {
outputChannel <- inrecAndContext // end-of-stream marker

24
go/src/miller/transforming/chain_transformer.go
View file

@ -48,8 +48,28 @@ func runSingleTransformer(
) {
for {
recordAndContext := <-inputChannel
recordTransformer.Transform(recordAndContext, outputChannel)
if recordAndContext.Record == nil { // end of stream
// Three things can come through:
//
// * End-of-stream marker
// * Non-nil records to be printed
// * Strings to be printed from put/filter DSL print/dump/etc
// statements. They are handled here 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.
//
// The first two are passed to the transformer. The third we send along
// the output channel without involving the record-transformer, since
// there is no record to be transformed.
if recordAndContext.EndOfStream == true || recordAndContext.Record != nil {
recordTransformer.Transform(recordAndContext, outputChannel)
} else {
outputChannel <- recordAndContext
}
if recordAndContext.EndOfStream {
break
}
}

54
go/src/miller/types/context.go
View file

@ -7,9 +7,18 @@ import (
// 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 *Mlrmap
Context Context
Record *Mlrmap
Context Context
OutputString string
EndOfStream bool
}
func NewRecordAndContext(
@ -21,7 +30,9 @@ func NewRecordAndContext(
// 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,
Context: *context,
OutputString: "",
EndOfStream: false,
}
}
@ -30,8 +41,33 @@ func (this *RecordAndContext) Copy() *RecordAndContext {
recordCopy := this.Record.Copy()
contextCopy := this.Context
return &RecordAndContext{
recordCopy,
contextCopy,
Record: recordCopy,
Context: contextCopy,
OutputString: "",
EndOfStream: false,
}
}
// For print/dump/etc to insert strings sequenced into the record-output stream.
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,
}
}
@ -106,11 +142,9 @@ func (this *Context) UpdateForStartOfFile(filename string) {
}
// For the record-readers to update their initial context as each new record is read.
func (this *Context) UpdateForInputRecord(inrec *Mlrmap) {
if inrec != nil { // do not count the end-of-stream marker which is a nil record pointer
this.NR++
this.FNR++
}
func (this *Context) UpdateForInputRecord() {
this.NR++
this.FNR++
}
func (this *Context) Copy() *Context {

13
go/todo.txt
View file

@ -2,12 +2,8 @@
TOP OF LIST:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
! emitx/io-redirect/etc
! fix all typemasks; 'int x = 1; x = "abc"'
! need to rethink concurrency of DSL-print and previous-record write ... :(
* I/O redirects
* merge-fields
* do some profiling every so often
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
@ -15,6 +11,13 @@ TOP OF LIST:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
emitp/emitf:
! ochan mod for all stdouts
o try non-nil record @ EOS for test
o need output-handler object ...
- stdout -- ochan <- x
- stderr -- out immediately
- write-to-file -- out immediately
- append-to-file -- out immediately
- write-to-pipe -- out immediately
* bug
run_mlr $input put $vflag '@x={"a":1}; @y={"b":2}; emit (@x, @y), "a"'
* new emitx punctuation-syntax -- decide x 4