From 3dd0b024196b4c762813684ca8bb81a342ccd7af Mon Sep 17 00:00:00 2001 From: John Kerl Date: Sun, 10 Jan 2021 12:11:02 -0500 Subject: [PATCH] end-of-stream nilrec -> bool refactor --- go/README.md | 8 +- go/src/miller/input/record_reader_csv.go | 7 +- go/src/miller/input/record_reader_csvlite.go | 9 +- go/src/miller/input/record_reader_dkvp.go | 7 +- go/src/miller/input/record_reader_json.go | 9 +- go/src/miller/input/record_reader_nidx.go | 7 +- go/src/miller/input/record_reader_xtab.go | 9 +- go/src/miller/output/channel_writer.go | 32 ++++- go/src/miller/transformers/altkv.go | 4 +- go/src/miller/transformers/bootstrap.go | 109 +++++++++--------- go/src/miller/transformers/cat.go | 8 +- go/src/miller/transformers/check.go | 2 +- go/src/miller/transformers/count-similar.go | 4 +- go/src/miller/transformers/count.go | 7 +- go/src/miller/transformers/cut.go | 12 +- go/src/miller/transformers/decimate.go | 4 +- go/src/miller/transformers/fill-down.go | 4 +- go/src/miller/transformers/flatten.go | 8 +- go/src/miller/transformers/gap.go | 8 +- go/src/miller/transformers/grep.go | 4 +- go/src/miller/transformers/group-by.go | 4 +- go/src/miller/transformers/group-like.go | 4 +- go/src/miller/transformers/head.go | 7 +- .../miller/transformers/join-bucket-keeper.go | 2 +- go/src/miller/transformers/join.go | 13 +-- go/src/miller/transformers/json-parse.go | 8 +- go/src/miller/transformers/json-stringify.go | 8 +- go/src/miller/transformers/label.go | 6 +- go/src/miller/transformers/nothing.go | 2 +- go/src/miller/transformers/put_or_filter.go | 8 +- go/src/miller/transformers/regularize.go | 4 +- .../transformers/remove-empty-columns.go | 4 +- go/src/miller/transformers/rename.go | 6 +- go/src/miller/transformers/reorder.go | 8 +- go/src/miller/transformers/sample.go | 4 +- go/src/miller/transformers/sec2gmt.go | 6 +- go/src/miller/transformers/seqgen.go | 5 +- go/src/miller/transformers/shuffle.go | 2 +- .../transformers/skip-trivial-records.go | 4 +- .../transformers/sort-within-records.go | 6 +- go/src/miller/transformers/sort.go | 4 +- go/src/miller/transformers/stats1.go | 2 +- go/src/miller/transformers/step.go | 6 +- go/src/miller/transformers/tac.go | 7 +- go/src/miller/transformers/tail.go | 4 +- go/src/miller/transformers/unflatten.go | 8 +- go/src/miller/transformers/unsparsify.go | 10 +- .../miller/transforming/chain_transformer.go | 24 +++- go/src/miller/types/context.go | 54 +++++++-- go/todo.txt | 13 ++- 50 files changed, 279 insertions(+), 226 deletions(-) diff --git a/go/README.md b/go/README.md index da89930a6..fd8a91a87 100644 --- a/go/README.md +++ b/go/README.md @@ -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 diff --git a/go/src/miller/input/record_reader_csv.go b/go/src/miller/input/record_reader_csv.go index 359da8182..a77ef32e2 100644 --- a/go/src/miller/input/record_reader_csv.go +++ b/go/src/miller/input/record_reader_csv.go @@ -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, diff --git a/go/src/miller/input/record_reader_csvlite.go b/go/src/miller/input/record_reader_csvlite.go index 40136df8a..27ec453af 100644 --- a/go/src/miller/input/record_reader_csvlite.go +++ b/go/src/miller/input/record_reader_csvlite.go @@ -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, diff --git a/go/src/miller/input/record_reader_dkvp.go b/go/src/miller/input/record_reader_dkvp.go index b7298eb60..787600fd0 100644 --- a/go/src/miller/input/record_reader_dkvp.go +++ b/go/src/miller/input/record_reader_dkvp.go @@ -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, diff --git a/go/src/miller/input/record_reader_json.go b/go/src/miller/input/record_reader_json.go index 5ea157431..3bbb7b6cd 100644 --- a/go/src/miller/input/record_reader_json.go +++ b/go/src/miller/input/record_reader_json.go @@ -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, diff --git a/go/src/miller/input/record_reader_nidx.go b/go/src/miller/input/record_reader_nidx.go index edfd9077c..ca8be2444 100644 --- a/go/src/miller/input/record_reader_nidx.go +++ b/go/src/miller/input/record_reader_nidx.go @@ -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, diff --git a/go/src/miller/input/record_reader_xtab.go b/go/src/miller/input/record_reader_xtab.go index f0fa3b178..08e48a2aa 100644 --- a/go/src/miller/input/record_reader_xtab.go +++ b/go/src/miller/input/record_reader_xtab.go @@ -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() } diff --git a/go/src/miller/output/channel_writer.go b/go/src/miller/output/channel_writer.go index 896b3e845..137e0ad40 100644 --- a/go/src/miller/output/channel_writer.go +++ b/go/src/miller/output/channel_writer.go @@ -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 } diff --git a/go/src/miller/transformers/altkv.go b/go/src/miller/transformers/altkv.go index 182000634..369da66fe 100644 --- a/go/src/miller/transformers/altkv.go +++ b/go/src/miller/transformers/altkv.go @@ -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 diff --git a/go/src/miller/transformers/bootstrap.go b/go/src/miller/transformers/bootstrap.go index ffbaad092..f2ca2a9c0 100644 --- a/go/src/miller/transformers/bootstrap.go +++ b/go/src/miller/transformers/bootstrap.go @@ -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 } diff --git a/go/src/miller/transformers/cat.go b/go/src/miller/transformers/cat.go index 7868b9fe8..ab1a60f3f 100644 --- a/go/src/miller/transformers/cat.go +++ b/go/src/miller/transformers/cat.go @@ -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 diff --git a/go/src/miller/transformers/check.go b/go/src/miller/transformers/check.go index 6bca23100..6c238be10 100644 --- a/go/src/miller/transformers/check.go +++ b/go/src/miller/transformers/check.go @@ -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 } } diff --git a/go/src/miller/transformers/count-similar.go b/go/src/miller/transformers/count-similar.go index 2ac2dec34..343003011 100644 --- a/go/src/miller/transformers/count-similar.go +++ b/go/src/miller/transformers/count-similar.go @@ -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 diff --git a/go/src/miller/transformers/count.go b/go/src/miller/transformers/count.go index 8e56013d5..e279b5ca1 100644 --- a/go/src/miller/transformers/count.go +++ b/go/src/miller/transformers/count.go @@ -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, diff --git a/go/src/miller/transformers/cut.go b/go/src/miller/transformers/cut.go index 25efaaf57..5e7f6a4d5 100644 --- a/go/src/miller/transformers/cut.go +++ b/go/src/miller/transformers/cut.go @@ -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) diff --git a/go/src/miller/transformers/decimate.go b/go/src/miller/transformers/decimate.go index 695146557..480ba695f 100644 --- a/go/src/miller/transformers/decimate.go +++ b/go/src/miller/transformers/decimate.go @@ -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 { diff --git a/go/src/miller/transformers/fill-down.go b/go/src/miller/transformers/fill-down.go index 14de73953..f2aa2dbea 100644 --- a/go/src/miller/transformers/fill-down.go +++ b/go/src/miller/transformers/fill-down.go @@ -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 diff --git a/go/src/miller/transformers/flatten.go b/go/src/miller/transformers/flatten.go index 8062c30ff..e4ba9d5cc 100644 --- a/go/src/miller/transformers/flatten.go +++ b/go/src/miller/transformers/flatten.go @@ -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 diff --git a/go/src/miller/transformers/gap.go b/go/src/miller/transformers/gap.go index aab396413..f9635ad17 100644 --- a/go/src/miller/transformers/gap.go +++ b/go/src/miller/transformers/gap.go @@ -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 { diff --git a/go/src/miller/transformers/grep.go b/go/src/miller/transformers/grep.go index 33e267010..f62468671 100644 --- a/go/src/miller/transformers/grep.go +++ b/go/src/miller/transformers/grep.go @@ -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 { diff --git a/go/src/miller/transformers/group-by.go b/go/src/miller/transformers/group-by.go index 2fc5f55c4..0037a5bdc 100644 --- a/go/src/miller/transformers/group-by.go +++ b/go/src/miller/transformers/group-by.go @@ -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 { diff --git a/go/src/miller/transformers/group-like.go b/go/src/miller/transformers/group-like.go index 4ca1b5a30..2dd44a143 100644 --- a/go/src/miller/transformers/group-like.go +++ b/go/src/miller/transformers/group-like.go @@ -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() diff --git a/go/src/miller/transformers/head.go b/go/src/miller/transformers/head.go index 81387c1a7..93bea70ef 100644 --- a/go/src/miller/transformers/head.go +++ b/go/src/miller/transformers/head.go @@ -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 { diff --git a/go/src/miller/transformers/join-bucket-keeper.go b/go/src/miller/transformers/join-bucket-keeper.go index acdd4fde4..05c4d5b63 100644 --- a/go/src/miller/transformers/join-bucket-keeper.go +++ b/go/src/miller/transformers/join-bucket-keeper.go @@ -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 { diff --git a/go/src/miller/transformers/join.go b/go/src/miller/transformers/join.go index 15f7573d5..19816f8e1 100644 --- a/go/src/miller/transformers/join.go +++ b/go/src/miller/transformers/join.go @@ -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, diff --git a/go/src/miller/transformers/json-parse.go b/go/src/miller/transformers/json-parse.go index 3d70f6c2a..c8a95405c 100644 --- a/go/src/miller/transformers/json-parse.go +++ b/go/src/miller/transformers/json-parse.go @@ -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() diff --git a/go/src/miller/transformers/json-stringify.go b/go/src/miller/transformers/json-stringify.go index ae46bc665..594e21aa9 100644 --- a/go/src/miller/transformers/json-stringify.go +++ b/go/src/miller/transformers/json-stringify.go @@ -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) diff --git a/go/src/miller/transformers/label.go b/go/src/miller/transformers/label.go index f9223a6a0..369d2dbb8 100644 --- a/go/src/miller/transformers/label.go +++ b/go/src/miller/transformers/label.go @@ -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 } diff --git a/go/src/miller/transformers/nothing.go b/go/src/miller/transformers/nothing.go index d52b3bb9e..16d510048 100644 --- a/go/src/miller/transformers/nothing.go +++ b/go/src/miller/transformers/nothing.go @@ -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 } } diff --git a/go/src/miller/transformers/put_or_filter.go b/go/src/miller/transformers/put_or_filter.go index a61ec1c0a..e7788bb7e 100644 --- a/go/src/miller/transformers/put_or_filter.go +++ b/go/src/miller/transformers/put_or_filter.go @@ -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) } } diff --git a/go/src/miller/transformers/regularize.go b/go/src/miller/transformers/regularize.go index 57e4fefa3..10feb0d2a 100644 --- a/go/src/miller/transformers/regularize.go +++ b/go/src/miller/transformers/regularize.go @@ -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, ",") diff --git a/go/src/miller/transformers/remove-empty-columns.go b/go/src/miller/transformers/remove-empty-columns.go index 1795009be..60b1f32f1 100644 --- a/go/src/miller/transformers/remove-empty-columns.go +++ b/go/src/miller/transformers/remove-empty-columns.go @@ -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 { diff --git a/go/src/miller/transformers/rename.go b/go/src/miller/transformers/rename.go index 8c3fef27b..9c091b7af 100644 --- a/go/src/miller/transformers/rename.go +++ b/go/src/miller/transformers/rename.go @@ -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 } diff --git a/go/src/miller/transformers/reorder.go b/go/src/miller/transformers/reorder.go index 30c525cb9..be31ab62d 100644 --- a/go/src/miller/transformers/reorder.go +++ b/go/src/miller/transformers/reorder.go @@ -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) } diff --git a/go/src/miller/transformers/sample.go b/go/src/miller/transformers/sample.go index ba0ee4412..511fa7ddc 100644 --- a/go/src/miller/transformers/sample.go +++ b/go/src/miller/transformers/sample.go @@ -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) diff --git a/go/src/miller/transformers/sec2gmt.go b/go/src/miller/transformers/sec2gmt.go index 061009f87..12fb92417 100644 --- a/go/src/miller/transformers/sec2gmt.go +++ b/go/src/miller/transformers/sec2gmt.go @@ -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 diff --git a/go/src/miller/transformers/seqgen.go b/go/src/miller/transformers/seqgen.go index 7638db31d..59f9b2bc6 100644 --- a/go/src/miller/transformers/seqgen.go +++ b/go/src/miller/transformers/seqgen.go @@ -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) } diff --git a/go/src/miller/transformers/shuffle.go b/go/src/miller/transformers/shuffle.go index ab2d2101d..380e1bd20 100644 --- a/go/src/miller/transformers/shuffle.go +++ b/go/src/miller/transformers/shuffle.go @@ -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 diff --git a/go/src/miller/transformers/skip-trivial-records.go b/go/src/miller/transformers/skip-trivial-records.go index 962388156..281968b93 100644 --- a/go/src/miller/transformers/skip-trivial-records.go +++ b/go/src/miller/transformers/skip-trivial-records.go @@ -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() != "" { diff --git a/go/src/miller/transformers/sort-within-records.go b/go/src/miller/transformers/sort-within-records.go index f8e83f1ee..26390a16a 100644 --- a/go/src/miller/transformers/sort-within-records.go +++ b/go/src/miller/transformers/sort-within-records.go @@ -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 } diff --git a/go/src/miller/transformers/sort.go b/go/src/miller/transformers/sort.go index d28c56d21..1e2145da7 100644 --- a/go/src/miller/transformers/sort.go +++ b/go/src/miller/transformers/sort.go @@ -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, diff --git a/go/src/miller/transformers/stats1.go b/go/src/miller/transformers/stats1.go index ed05b31b1..a08f5e00d 100644 --- a/go/src/miller/transformers/stats1.go +++ b/go/src/miller/transformers/stats1.go @@ -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) diff --git a/go/src/miller/transformers/step.go b/go/src/miller/transformers/step.go index 1b0a07137..d4f637b5d 100644 --- a/go/src/miller/transformers/step.go +++ b/go/src/miller/transformers/step.go @@ -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" diff --git a/go/src/miller/transformers/tac.go b/go/src/miller/transformers/tac.go index bdaf3c988..3a3527c94 100644 --- a/go/src/miller/transformers/tac.go +++ b/go/src/miller/transformers/tac.go @@ -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) } } diff --git a/go/src/miller/transformers/tail.go b/go/src/miller/transformers/tail.go index 68fff6008..3fd88a6e5 100644 --- a/go/src/miller/transformers/tail.go +++ b/go/src/miller/transformers/tail.go @@ -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 { diff --git a/go/src/miller/transformers/unflatten.go b/go/src/miller/transformers/unflatten.go index 4bab782fe..f92cedf68 100644 --- a/go/src/miller/transformers/unflatten.go +++ b/go/src/miller/transformers/unflatten.go @@ -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 diff --git a/go/src/miller/transformers/unsparsify.go b/go/src/miller/transformers/unsparsify.go index 4aacd5b22..edbee8b7b 100644 --- a/go/src/miller/transformers/unsparsify.go +++ b/go/src/miller/transformers/unsparsify.go @@ -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 diff --git a/go/src/miller/transforming/chain_transformer.go b/go/src/miller/transforming/chain_transformer.go index 57e66dba6..b5d8b2bd3 100644 --- a/go/src/miller/transforming/chain_transformer.go +++ b/go/src/miller/transforming/chain_transformer.go @@ -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 } } diff --git a/go/src/miller/types/context.go b/go/src/miller/types/context.go index 2d7d3efdc..74c3f2100 100644 --- a/go/src/miller/types/context.go +++ b/go/src/miller/types/context.go @@ -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 { diff --git a/go/todo.txt b/go/todo.txt index 1a993b597..f131a9cce 100644 --- a/go/todo.txt +++ b/go/todo.txt @@ -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