Mirrors/miller
1
0
Fork 0
mirror of https://github.com/johnkerl/miller.git synced 2026-07-17 16:38:54 +00:00
Code Issues Projects Releases Wiki Activity

Remove os.Exit callsites below the entrypoint: phase 3 (plans/exit.md) (#2204)

Browse source 
Phase 3 of plans/exit.md: the streaming-interface change, the load-bearing
piece for #341 (DSL exit statement) and #440 (strict mode).

- RecordTransformer.Transform and RecordTransformerFunc now return error.
  All 69 Transform implementations and their dispatch helpers updated
  (mechanical rewrite, compiler- and errcheck-verified).
- runSingleTransformerBatch, on a Transform error, forwards any output
  produced before the failure plus an end-of-stream marker downstream, so
  the rest of the chain and the record-writer drain and finish cleanly;
  runSingleTransformer then surfaces the error to stream.Stream's select
  loop (non-blocking send; first error wins) and signals upstream-done so
  the record-reader stops. This is exactly the flush-then-exit sequencing a
  future DSL 'exit N' needs.
- dataProcessingErrorChannel and FileOutputHandler.recordErroredChannel are
  now chan error instead of chan bool; ChannelWriter still prints write-
  error details at the site and sends the 'exiting due to data error'
  sentinel, preserving the exact stderr shape pinned by regression cases.
- Mid-stream os.Exit sites converted to returned errors: put/filter DSL
  begin/main/end-block errors and the non-boolean filter-expression case,
  tee write/close failures, split write/open/close failures, join left-file
  ingest failures (both half-streaming and sorted paths, with full error
  plumbing through JoinBucketKeeper), histogram/stats2 ingest errors, surv
  fit errors, and step stepper allocation (tStepperAllocator now returns
  (tStepper, error); bad EWMA coefficients propagate; negative slwin
  parameters are reported by the CLI parser via the existing
  bad-stepper-name pattern).
- The two genuinely internal join-bucket-keeper states now use
  lib.InternalCodingErrorWithMessageIf instead of hand-rolled print+exit.
- pkg/transformers is now os.Exit-free.

Behavior notes: the non-boolean filter message gains the standard 'mlr: '
prefix and a newline (it previously printed with neither); tee errors now
include the underlying cause. All 4779 regression cases pass unchanged;
mlr head early-out latency is unaffected (0.02s over 50M records).

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
John Kerl 2026-07-15 15:07:09 -04:00 • committed by GitHub
parent f17276e0c4
commit 570fcf0de4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
78 changed files with 604 additions and 405 deletions
Download patch file Download diff file Expand all files Collapse all files

18
pkg/output/channel_writer.go
View file

@ -2,6 +2,7 @@ package output
import (
"bufio"
"errors"
"fmt"
"os"
@ -14,7 +15,7 @@ func ChannelWriter(
recordWriter IRecordWriter,
writerOptions *cli.TWriterOptions,
doneChannel chan<- bool,
dataProcessingErrorChannel chan<- bool,
dataProcessingErrorChannel chan<- error,
bufferedOutputStream *bufio.Writer,
outputIsStdout bool,
) {
@ -25,12 +26,17 @@ func ChannelWriter(
recordsAndContexts,
recordWriter,
writerOptions,
dataProcessingErrorChannel,
bufferedOutputStream,
outputIsStdout,
)
if errored {
dataProcessingErrorChannel <- true
// Details have already been printed to stderr by
// channelWriterHandleBatch. Non-blocking send: if a transformer
// errored first, that error wins.
select {
case dataProcessingErrorChannel <- errors.New("exiting due to data error"):
default:
}
doneChannel <- true
break
}
@ -41,13 +47,13 @@ func ChannelWriter(
}
}
// TODO: comment
// Returns true on end of record stream
// channelWriterHandleBatch writes one batch of records. The first return
// value is true on end of record stream; the second is true on a data error
// (details already printed to stderr here).
func channelWriterHandleBatch(
recordsAndContexts []*types.RecordAndContext,
recordWriter IRecordWriter,
writerOptions *cli.TWriterOptions,
dataProcessingErrorChannel chan<- bool,
bufferedOutputStream *bufio.Writer,
outputIsStdout bool,
) (done bool, errored bool) {

9
pkg/output/file_output_handlers.go
View file

@ -13,7 +13,6 @@ package output
import (
"bufio"
"errors"
"fmt"
"io"
"os"
@ -313,7 +312,7 @@ type FileOutputHandler struct {
recordWriter IRecordWriter
recordOutputChannel chan []*types.RecordAndContext // list of *types.RecordAndContext
recordDoneChannel chan bool
recordErroredChannel chan bool
recordErroredChannel chan error
}
func newOutputHandlerCommon(
@ -460,7 +459,7 @@ func (handler *FileOutputHandler) setUpRecordWriter() error {
handler.recordOutputChannel = make(chan []*types.RecordAndContext, 1) // list of *types.RecordAndContext
handler.recordDoneChannel = make(chan bool, 1)
handler.recordErroredChannel = make(chan bool, 1)
handler.recordErroredChannel = make(chan error, 1)
go ChannelWriter(
handler.recordOutputChannel,
@ -487,9 +486,9 @@ func (handler *FileOutputHandler) Close() (retval error) {
done := false
for !done {
select {
case <-handler.recordErroredChannel:
case werr := <-handler.recordErroredChannel:
done = true
retval = errors.New("exiting due to data error") // details already printed
retval = werr // details already printed
case <-handler.recordDoneChannel:
done = true
}

15
pkg/stream/stream.go
View file

@ -2,7 +2,6 @@ package stream
import (
"bufio"
"errors"
"io"
"github.com/johnkerl/miller/v6/pkg/cli"
@ -66,10 +65,14 @@ func Stream(
// We're done when a fatal error is registered on input (file not found,
// etc) or when the record-writer has written all its output. We use
// channels to communicate both of these conditions.
// channels to communicate both of these conditions. The
// dataProcessingErrorChannel carries mid-stream errors from the
// transformer chain (e.g. DSL runtime errors, tee/split write failures)
// and from the record-writer; senders use a non-blocking send, so the
// first error wins and is returned once the writer finishes draining.
inputErrorChannel := make(chan error, 1)
doneWritingChannel := make(chan bool, 1)
dataProcessingErrorChannel := make(chan bool, 1)
dataProcessingErrorChannel := make(chan error, 1)
// For mlr head, so a transformer can communicate it will disregard all
// further input. It writes this back upstream, and that is passed back to
@ -84,7 +87,7 @@ func Stream(
go recordReader.Read(fileNames, *initialContext, readerChannel, inputErrorChannel, readerDownstreamDoneChannel)
go transformers.ChainTransformer(readerChannel, readerDownstreamDoneChannel, recordTransformers,
writerChannel, options)
writerChannel, dataProcessingErrorChannel, options)
go output.ChannelWriter(writerChannel, recordWriter, &options.WriterOptions, doneWritingChannel,
dataProcessingErrorChannel, bufferedOutputStream, outputIsStdout)
@ -94,8 +97,8 @@ func Stream(
select {
case ierr := <-inputErrorChannel:
retval = ierr
case <-dataProcessingErrorChannel:
retval = errors.New("exiting due to data error") // details already printed
case derr := <-dataProcessingErrorChannel:
retval = derr
case <-doneWritingChannel:
done = true
}

46
pkg/transformers/aaa_chain_transformer.go
View file

@ -140,6 +140,7 @@ func ChainTransformer(
readerDownstreamDoneChannel chan<- bool, // for mlr head -- see also stream.go
recordTransformers []RecordTransformer, // not *recordTransformer since this is an interface
writerRecordChannel chan<- []*types.RecordAndContext, // list of *types.RecordAndContext
dataProcessingErrorChannel chan<- error, // for mid-stream transformer errors -- see stream.go
options *cli.TOptions,
) {
i := 0
@ -184,6 +185,7 @@ func ChainTransformer(
orchan,
idchan,
odchan,
dataProcessingErrorChannel,
options,
)
}
@ -196,13 +198,15 @@ func runSingleTransformer(
outputRecordChannel chan<- []*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
dataProcessingErrorChannel chan<- error,
options *cli.TOptions,
) {
done := false
for !done {
recordsAndContexts := <-inputRecordChannel
done = runSingleTransformerBatch(
var err error
done, err = runSingleTransformerBatch(
recordsAndContexts,
recordTransformer,
isFirstInChain,
@ -211,11 +215,33 @@ func runSingleTransformer(
outputDownstreamDoneChannel,
options,
)
if err != nil {
// Surface the error to stream.Stream's select loop. Non-blocking
// send: if another goroutine errored first, that error wins.
select {
case dataProcessingErrorChannel <- err:
default:
}
// Tell upstream (transformers and ultimately the record-reader,
// via the mlr-head mechanism) that we'll ignore further input.
select {
case outputDownstreamDoneChannel <- true:
default:
}
// runSingleTransformerBatch has already forwarded an end-of-stream
// marker downstream, so the record-writer drains and finishes,
// upon which stream.Stream returns the error we sent above.
return
}
}
}
// TODO: comment
// Returns true on end of record stream
// runSingleTransformerBatch passes one batch of records through the
// transformer. The boolean return is true on end of record stream. A non-nil
// error is a mid-stream transformer failure: any output produced before the
// failure, plus an end-of-stream marker, has already been forwarded
// downstream so the rest of the chain and the record-writer can drain and
// finish cleanly.
func runSingleTransformerBatch(
inputRecordsAndContexts []*types.RecordAndContext, // list of types.RecordAndContext
recordTransformer RecordTransformer,
@ -224,7 +250,7 @@ func runSingleTransformerBatch(
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
options *cli.TOptions,
) bool {
) (bool, error) {
outputRecordsAndContexts := make([]*types.RecordAndContext, 0, len(inputRecordsAndContexts))
done := false
@ -257,7 +283,7 @@ func runSingleTransformerBatch(
// there is no record to be transformed.
if inputRecordAndContext.EndOfStream || inputRecordAndContext.Record != nil {
recordTransformer.Transform(
err := recordTransformer.Transform(
inputRecordAndContext,
&outputRecordsAndContexts,
// TODO: maybe refactor these out of each transformer.
@ -265,6 +291,14 @@ func runSingleTransformerBatch(
inputDownstreamDoneChannel,
outputDownstreamDoneChannel,
)
if err != nil {
// Forward what was produced before the failure, plus an
// end-of-stream marker so downstream drains and finishes.
outputRecordsAndContexts = append(outputRecordsAndContexts,
types.NewEndOfStreamMarker(&inputRecordAndContext.Context))
outputRecordChannel <- outputRecordsAndContexts
return true, err
}
} else {
outputRecordsAndContexts = append(outputRecordsAndContexts, inputRecordAndContext)
}
@ -281,5 +315,5 @@ func runSingleTransformerBatch(
outputRecordChannel <- outputRecordsAndContexts
return done
return done, nil
}

11
pkg/transformers/aaa_record_transformer.go
View file

@ -10,13 +10,20 @@ import (
// RecordTransformer is the interface satisfied by all transformers, i.e.,
// Miller verbs. See stream.go for context on the channels, as well as for
// context on end-of-record-stream signaling.
//
// A non-nil error from Transform means a mid-stream failure (e.g. a DSL
// runtime error, or a tee/split output file that can't be written). The
// chain driver forwards any already-produced output plus an end-of-stream
// marker downstream so the record-writer drains and finishes, then surfaces
// the error up through stream.Stream to the entrypoint layer (see
// plans/exit.md). Transformers must not call os.Exit.
type RecordTransformer interface {
Transform(
inrecAndContext *types.RecordAndContext,
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
)
) error
}
type RecordTransformerFunc func(
@ -24,7 +31,7 @@ type RecordTransformerFunc func(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
)
) error
// Used within some verbs
type RecordTransformerHelperFunc func(

3
pkg/transformers/altkv.go
View file

@ -87,7 +87,7 @@ func (tr *TransformerAltkv) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -120,4 +120,5 @@ func (tr *TransformerAltkv) Transform(
} else { // end of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

12
pkg/transformers/bar.go
View file

@ -226,9 +226,9 @@ func (tr *TransformerBar) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerBar) processNoAuto(
@ -236,7 +236,7 @@ func (tr *TransformerBar) processNoAuto(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -257,6 +257,7 @@ func (tr *TransformerBar) processNoAuto(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerBar) processAuto(
@ -264,10 +265,10 @@ func (tr *TransformerBar) processAuto(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
tr.recordsForAutoMode = append(tr.recordsForAutoMode, inrecAndContext.Copy())
return
return nil
}
// Else, end of stream
@ -336,4 +337,5 @@ func (tr *TransformerBar) processAuto(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, tr.recordsForAutoMode...)
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // Emit the end-of-stream marker
return nil
}

7
pkg/transformers/bootstrap.go
View file

@ -108,12 +108,12 @@ func (tr *TransformerBootstrap) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
// Not end of input stream: retain the record, and emit nothing until end of stream.
if !inrecAndContext.EndOfStream {
tr.recordsAndContexts = append(tr.recordsAndContexts, inrecAndContext)
return
return nil
}
// Else end of record stream
@ -148,7 +148,7 @@ func (tr *TransformerBootstrap) Transform(
if nout == 0 {
// Emit the stream-terminating null record
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
// Make an array of pointers into the input list.
@ -165,4 +165,5 @@ func (tr *TransformerBootstrap) Transform(
// Emit the stream-terminating null record
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}

3
pkg/transformers/bootstrap_ci.go
View file

@ -217,13 +217,14 @@ func (tr *TransformerBootstrapCI) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.handleInputRecord(inrecAndContext)
} else {
tr.handleEndOfRecordStream(inrecAndContext, outputRecordsAndContexts)
}
return nil
}
func (tr *TransformerBootstrapCI) handleInputRecord(

4
pkg/transformers/bootstrap_ci_test.go
View file

@ -57,14 +57,14 @@ func feedBootstrapCI(tr *TransformerBootstrapCI, xValues []int64) []*types.Recor
record := mlrval.NewMlrmapAsRecord()
record.PutCopy("g", mlrval.FromString("one"))
record.PutCopy("x", mlrval.FromInt(xValue))
tr.Transform(
_ = tr.Transform(
types.NewRecordAndContext(record, context),
&outputRecordsAndContexts,
inputDownstreamDoneChannel,
outputDownstreamDoneChannel,
)
}
tr.Transform(
_ = tr.Transform(
types.NewEndOfStreamMarker(context),
&outputRecordsAndContexts,
inputDownstreamDoneChannel,

14
pkg/transformers/case.go
View file

@ -184,10 +184,10 @@ func (tr *TransformerCase) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.recordTransformerFunc(
return tr.recordTransformerFunc(
inrecAndContext,
outputRecordsAndContexts,
inputDownstreamDoneChannel,
@ -196,6 +196,7 @@ func (tr *TransformerCase) Transform(
} else { // end of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerCase) transformKeysOnly(
@ -203,7 +204,7 @@ func (tr *TransformerCase) transformKeysOnly(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
_ <-chan bool,
__ chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
newrec := mlrval.NewMlrmapAsRecord()
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -216,6 +217,7 @@ func (tr *TransformerCase) transformKeysOnly(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(newrec, &inrecAndContext.Context))
return nil
}
func (tr *TransformerCase) transformValuesOnly(
@ -223,7 +225,7 @@ func (tr *TransformerCase) transformValuesOnly(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
_ <-chan bool,
__ chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if tr.fieldNameSet == nil || tr.fieldNameSet[pe.Key] {
@ -234,6 +236,7 @@ func (tr *TransformerCase) transformValuesOnly(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(inrec, &inrecAndContext.Context))
return nil
}
func (tr *TransformerCase) transformKeysAndValues(
@ -241,7 +244,7 @@ func (tr *TransformerCase) transformKeysAndValues(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
_ <-chan bool,
__ chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
newrec := mlrval.NewMlrmapAsRecord()
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -259,4 +262,5 @@ func (tr *TransformerCase) transformKeysAndValues(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(newrec, &inrecAndContext.Context))
return nil
}

13
pkg/transformers/cat.go
View file

@ -171,9 +171,9 @@ func (tr *TransformerCat) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(
return tr.recordTransformerFunc(
inrecAndContext,
outputRecordsAndContexts,
inputDownstreamDoneChannel,
@ -186,7 +186,7 @@ func (tr *TransformerCat) simpleCat(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
if tr.doFileName {
inrecAndContext.Record.PrependCopy("filename", mlrval.FromString(inrecAndContext.Context.FILENAME))
@ -196,6 +196,7 @@ func (tr *TransformerCat) simpleCat(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
func (tr *TransformerCat) countersUngrouped(
@ -203,7 +204,7 @@ func (tr *TransformerCat) countersUngrouped(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
tr.counter++
@ -218,6 +219,7 @@ func (tr *TransformerCat) countersUngrouped(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
func (tr *TransformerCat) countersGrouped(
@ -225,7 +227,7 @@ func (tr *TransformerCat) countersGrouped(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -256,4 +258,5 @@ func (tr *TransformerCat) countersGrouped(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}

3
pkg/transformers/check.go
View file

@ -93,7 +93,7 @@ func (tr *TransformerCheck) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -118,4 +118,5 @@ func (tr *TransformerCheck) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

13
pkg/transformers/clean_whitespace.go
View file

@ -130,9 +130,9 @@ func (tr *TransformerCleanWhitespace) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerCleanWhitespace) cleanWhitespaceInKeysAndValues(
@ -140,7 +140,7 @@ func (tr *TransformerCleanWhitespace) cleanWhitespaceInKeysAndValues(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
newrec := mlrval.NewMlrmapAsRecord()
@ -156,6 +156,7 @@ func (tr *TransformerCleanWhitespace) cleanWhitespaceInKeysAndValues(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerCleanWhitespace) cleanWhitespaceInKeys(
@ -163,7 +164,7 @@ func (tr *TransformerCleanWhitespace) cleanWhitespaceInKeys(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
newrec := mlrval.NewMlrmapAsRecord()
@ -178,6 +179,7 @@ func (tr *TransformerCleanWhitespace) cleanWhitespaceInKeys(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerCleanWhitespace) cleanWhitespaceInValues(
@ -185,7 +187,7 @@ func (tr *TransformerCleanWhitespace) cleanWhitespaceInValues(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
for pe := inrecAndContext.Record.Head; pe != nil; pe = pe.Next {
pe.Value = bifs.BIF_clean_whitespace(pe.Value)
@ -194,4 +196,5 @@ func (tr *TransformerCleanWhitespace) cleanWhitespaceInValues(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

12
pkg/transformers/count.go
View file

@ -157,9 +157,9 @@ func (tr *TransformerCount) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerCount) countUngrouped(
@ -167,7 +167,7 @@ func (tr *TransformerCount) countUngrouped(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
tr.ungroupedCount++
} else {
@ -177,6 +177,7 @@ func (tr *TransformerCount) countUngrouped(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerCount) countGrouped(
@ -184,7 +185,7 @@ func (tr *TransformerCount) countGrouped(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -192,7 +193,7 @@ func (tr *TransformerCount) countGrouped(
tr.groupByFieldNames,
)
if !ok { // Current record does not have specified fields; ignore
return
return nil
}
if !tr.groupedCounts.Has(groupingKey) {
@ -243,4 +244,5 @@ func (tr *TransformerCount) countGrouped(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

5
pkg/transformers/count_similar.go
View file

@ -130,14 +130,14 @@ func (tr *TransformerCountSimilar) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok { // This particular record doesn't have the specified fields; ignore
return
return nil
}
recordListForGroup := tr.recordListsByGroup.Get(groupingKey)
@ -164,4 +164,5 @@ func (tr *TransformerCountSimilar) Transform(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // Emit the stream-terminating null record
}
return nil
}

16
pkg/transformers/cut.go
View file

@ -177,9 +177,9 @@ func (tr *TransformerCut) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
// mlr cut -f a,b,c
@ -188,7 +188,7 @@ func (tr *TransformerCut) includeWithInputOrder(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
outrec := mlrval.NewMlrmap()
@ -204,6 +204,7 @@ func (tr *TransformerCut) includeWithInputOrder(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
// mlr cut -o -f a,b,c
@ -212,7 +213,7 @@ func (tr *TransformerCut) includeWithArgOrder(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
outrec := mlrval.NewMlrmap()
@ -227,6 +228,7 @@ func (tr *TransformerCut) includeWithArgOrder(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
// mlr cut -x -f a,b,c
@ -235,7 +237,7 @@ func (tr *TransformerCut) exclude(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for _, fieldName := range tr.fieldNameList {
@ -245,6 +247,7 @@ func (tr *TransformerCut) exclude(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
type entryIndex struct {
@ -257,7 +260,7 @@ func (tr *TransformerCut) processWithRegexes(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
newrec := mlrval.NewMlrmapAsRecord()
@ -295,4 +298,5 @@ func (tr *TransformerCut) processWithRegexes(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

5
pkg/transformers/decimate.go
View file

@ -147,14 +147,14 @@ func (tr *TransformerDecimate) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok {
return // This particular record doesn't have the specified fields; ignore
return nil // This particular record doesn't have the specified fields; ignore
}
countForGroup, ok := tr.countsByGroup[groupingKey]
@ -174,4 +174,5 @@ func (tr *TransformerDecimate) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // Emit the stream-terminating null record
}
return nil
}

3
pkg/transformers/describe.go
View file

@ -174,13 +174,14 @@ func (tr *TransformerDescribe) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.ingest(inrecAndContext)
} else {
tr.emit(inrecAndContext, outputRecordsAndContexts)
}
return nil
}
func (tr *TransformerDescribe) ingest(

10
pkg/transformers/fill_down.go
View file

@ -145,9 +145,9 @@ func (tr *TransformerFillDown) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerFillDown) transformSpecified(
@ -155,7 +155,7 @@ func (tr *TransformerFillDown) transformSpecified(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -185,6 +185,7 @@ func (tr *TransformerFillDown) transformSpecified(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerFillDown) transformAll(
@ -192,7 +193,7 @@ func (tr *TransformerFillDown) transformAll(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -223,4 +224,5 @@ func (tr *TransformerFillDown) transformAll(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

3
pkg/transformers/fill_empty.go
View file

@ -115,7 +115,7 @@ func (tr *TransformerFillEmpty) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -131,4 +131,5 @@ func (tr *TransformerFillEmpty) Transform(
} else { // end of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}

10
pkg/transformers/flatten.go
View file

@ -141,9 +141,9 @@ func (tr *TransformerFlatten) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerFlatten) flattenAll(
@ -151,7 +151,7 @@ func (tr *TransformerFlatten) flattenAll(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
oFlatSep := tr.oFlatSep
@ -163,6 +163,7 @@ func (tr *TransformerFlatten) flattenAll(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerFlatten) flattenSome(
@ -170,7 +171,7 @@ func (tr *TransformerFlatten) flattenSome(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
oFlatSep := tr.oFlatSep
@ -182,4 +183,5 @@ func (tr *TransformerFlatten) flattenSome(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

5
pkg/transformers/format_values.go
View file

@ -166,11 +166,11 @@ func (tr *TransformerFormatValues) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if inrecAndContext.EndOfStream {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
return
return nil
}
for pe := inrecAndContext.Record.Head; pe != nil; pe = pe.Next {
@ -195,4 +195,5 @@ func (tr *TransformerFormatValues) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}

3
pkg/transformers/fraction.go
View file

@ -193,7 +193,7 @@ func (tr *TransformerFraction) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream { // Not end of stream; pass 1
inrec := inrecAndContext.Record
@ -289,4 +289,5 @@ func (tr *TransformerFraction) Transform(
tr.recordsAndContexts = tr.recordsAndContexts[:0]
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

10
pkg/transformers/gap.go
View file

@ -142,9 +142,9 @@ func (tr *TransformerGap) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerGap) transformUnkeyed(
@ -152,7 +152,7 @@ func (tr *TransformerGap) transformUnkeyed(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
if tr.recordCount > 0 && tr.recordCount%tr.gapCount == 0 {
newrec := mlrval.NewMlrmapAsRecord()
@ -165,6 +165,7 @@ func (tr *TransformerGap) transformUnkeyed(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerGap) transformKeyed(
@ -172,7 +173,7 @@ func (tr *TransformerGap) transformKeyed(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -194,4 +195,5 @@ func (tr *TransformerGap) transformKeyed(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

3
pkg/transformers/grep.go
View file

@ -152,7 +152,7 @@ func (tr *TransformerGrep) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -175,4 +175,5 @@ func (tr *TransformerGrep) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

5
pkg/transformers/group_by.go
View file

@ -109,14 +109,14 @@ func (tr *TransformerGroupBy) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok {
return
return nil
}
recordListForGroup := tr.recordListsByGroup.Get(groupingKey)
@ -134,4 +134,5 @@ func (tr *TransformerGroupBy) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

3
pkg/transformers/group_like.go
View file

@ -92,7 +92,7 @@ func (tr *TransformerGroupLike) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -114,4 +114,5 @@ func (tr *TransformerGroupLike) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

36
pkg/transformers/having_fields.go
View file

@ -231,9 +231,9 @@ func (tr *TransformerHavingFields) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerHavingFields) transformHavingFieldsAtLeast(
@ -241,7 +241,7 @@ func (tr *TransformerHavingFields) transformHavingFieldsAtLeast(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
numFound := int64(0)
@ -250,7 +250,7 @@ func (tr *TransformerHavingFields) transformHavingFieldsAtLeast(
numFound++
if numFound == tr.numFieldNames {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
}
}
@ -258,6 +258,7 @@ func (tr *TransformerHavingFields) transformHavingFieldsAtLeast(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerHavingFields) transformHavingFieldsWhichAre(
@ -265,21 +266,22 @@ func (tr *TransformerHavingFields) transformHavingFieldsWhichAre(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
if inrec.FieldCount != tr.numFieldNames {
return
return nil
}
for pe := inrec.Head; pe != nil; pe = pe.Next {
if !tr.fieldNameSet[pe.Key] {
return
return nil
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerHavingFields) transformHavingFieldsAtMost(
@ -287,18 +289,19 @@ func (tr *TransformerHavingFields) transformHavingFieldsAtMost(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if !tr.fieldNameSet[pe.Key] {
return
return nil
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerHavingFields) transformHavingAllFieldsMatching(
@ -306,18 +309,19 @@ func (tr *TransformerHavingFields) transformHavingAllFieldsMatching(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if !tr.regex.MatchString(pe.Key) {
return
return nil
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerHavingFields) transformHavingAnyFieldsMatching(
@ -325,18 +329,19 @@ func (tr *TransformerHavingFields) transformHavingAnyFieldsMatching(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if tr.regex.MatchString(pe.Key) {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
}
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerHavingFields) transformHavingNoFieldsMatching(
@ -344,16 +349,17 @@ func (tr *TransformerHavingFields) transformHavingNoFieldsMatching(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
if tr.regex.MatchString(pe.Key) {
return
return nil
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

17
pkg/transformers/head.go
View file

@ -150,9 +150,9 @@ func (tr *TransformerHead) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerHead) transformUnkeyed(
@ -160,7 +160,7 @@ func (tr *TransformerHead) transformUnkeyed(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
tr.unkeyedRecordCount++
if tr.unkeyedRecordCount <= tr.headCount {
@ -176,6 +176,7 @@ func (tr *TransformerHead) transformUnkeyed(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerHead) transformKeyed(
@ -183,13 +184,13 @@ func (tr *TransformerHead) transformKeyed(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok {
return
return nil
}
count, present := tr.keyedRecordCounts[groupingKey]
@ -208,6 +209,7 @@ func (tr *TransformerHead) transformKeyed(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerHead) transformAllButLast(
@ -215,13 +217,13 @@ func (tr *TransformerHead) transformAllButLast(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok {
return
return nil
}
recordListForGroup := tr.recordListsByGroup[groupingKey]
@ -241,4 +243,5 @@ func (tr *TransformerHead) transformAllButLast(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

25
pkg/transformers/histogram.go
View file

@ -222,9 +222,9 @@ func (tr *TransformerHistogram) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerHistogram) transformNonAuto(
@ -232,30 +232,29 @@ func (tr *TransformerHistogram) transformNonAuto(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
tr.ingestNonAuto(inrecAndContext)
if err := tr.ingestNonAuto(inrecAndContext); err != nil {
return err
}
} else {
tr.emitNonAuto(&inrecAndContext.Context, outputRecordsAndContexts)
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerHistogram) ingestNonAuto(
inrecAndContext *types.RecordAndContext,
) {
) error {
inrec := inrecAndContext.Record
for _, valueFieldName := range tr.valueFieldNames {
stringValue := inrec.Get(valueFieldName)
if stringValue != nil {
floatValue, ok := stringValue.GetNumericToFloatValue()
if !ok {
fmt.Fprintf(
os.Stderr,
"%s %s: cannot parse \"%s\" as float.\n",
"mlr", verbNameHistogram, stringValue.String(),
)
os.Exit(1)
return cli.VerbErrorf(verbNameHistogram,
"cannot parse \"%s\" as float.", stringValue.String())
}
if (floatValue >= tr.lo) && (floatValue < tr.hi) {
idx := int((floatValue - tr.lo) * tr.mul)
@ -266,6 +265,7 @@ func (tr *TransformerHistogram) ingestNonAuto(
}
}
}
return nil
}
// sparklineRecord summarizes a field's binned counts as a single record with
@ -334,13 +334,14 @@ func (tr *TransformerHistogram) transformAuto(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
tr.ingestAuto(inrecAndContext)
} else {
tr.emitAuto(&inrecAndContext.Context, outputRecordsAndContexts)
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerHistogram) ingestAuto(

42
pkg/transformers/join.go
View file

@ -346,13 +346,17 @@ func NewTransformerJoin(
// miss anything. This lets people do joins that would otherwise take
// too much RAM.
tr.joinBucketKeeper = utils.NewJoinBucketKeeper(
joinBucketKeeper, err := utils.NewJoinBucketKeeper(
// opts.prepipe,
opts.leftFileName,
&opts.joinFlagOptions.ReaderOptions,
opts.leftJoinFieldNames,
tr.leftKeepFieldNameSet,
)
if err != nil {
return nil, err
}
tr.joinBucketKeeper = joinBucketKeeper
tr.recordTransformerFunc = tr.transformDoublyStreaming
}
@ -365,9 +369,9 @@ func (tr *TransformerJoin) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts,
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts,
inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
@ -378,14 +382,16 @@ func (tr *TransformerJoin) transformHalfStreaming(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
// This can't be done in the CLI-parser since it requires information which
// isn't known until after the CLI-parser is called.
//
// TODO: check if this is still true for the Go port, once everything else
// is done.
if !tr.ingested { // First call
tr.ingestLeftFile()
if err := tr.ingestLeftFile(); err != nil {
return err
}
tr.ingested = true
}
@ -423,6 +429,7 @@ func (tr *TransformerJoin) transformHalfStreaming(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerJoin) transformDoublyStreaming(
@ -430,7 +437,7 @@ func (tr *TransformerJoin) transformDoublyStreaming(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
keeper := tr.joinBucketKeeper // keystroke-saver
if !rightRecAndContext.EndOfStream {
@ -441,7 +448,11 @@ func (tr *TransformerJoin) transformDoublyStreaming(
tr.opts.rightJoinFieldNames,
)
if hasAllJoinKeys {
isPaired = keeper.FindJoinBucket(rightFieldValues)
var err error
isPaired, err = keeper.FindJoinBucket(rightFieldValues)
if err != nil {
return err
}
}
if tr.opts.emitLeftUnpairables {
tr.outputLeftUnpaireds(keeper, outputRecordsAndContexts)
@ -461,7 +472,9 @@ func (tr *TransformerJoin) transformDoublyStreaming(
}
} else { // end of record stream
keeper.FindJoinBucket(nil)
if _, err := keeper.FindJoinBucket(nil); err != nil {
return err
}
if tr.opts.emitLeftUnpairables {
tr.outputLeftUnpaireds(keeper, outputRecordsAndContexts)
@ -469,6 +482,7 @@ func (tr *TransformerJoin) transformDoublyStreaming(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, rightRecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerJoin) outputLeftUnpaireds(
@ -488,15 +502,14 @@ func (tr *TransformerJoin) outputLeftUnpaireds(
// Note: this logic is very similar to that in stream.go, which is what
// processes the main/right files.
func (tr *TransformerJoin) ingestLeftFile() {
func (tr *TransformerJoin) ingestLeftFile() error {
readerOpts := &tr.opts.joinFlagOptions.ReaderOptions
// Instantiate the record-reader
// TODO: perhaps increase recordsPerBatch, and/or refactor
recordReader, err := input.Create(readerOpts, 1)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
// Set the initial context for the left-file.
@ -525,8 +538,7 @@ func (tr *TransformerJoin) ingestLeftFile() {
select {
case err := <-errorChannel:
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
case leftrecsAndContexts := <-readerChannel:
// TODO: temp for batch-reader refactor
@ -542,8 +554,7 @@ func (tr *TransformerJoin) ingestLeftFile() {
// before declaring the ingest complete.
select {
case err := <-errorChannel:
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
default:
}
done = true
@ -572,6 +583,7 @@ func (tr *TransformerJoin) ingestLeftFile() {
}
}
}
return nil
}
// This helper method is used by the half-streaming/unsorted join, as well as

10
pkg/transformers/json_parse.go
View file

@ -143,9 +143,9 @@ func (tr *TransformerJSONParse) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerJSONParse) jsonParseAll(
@ -153,7 +153,7 @@ func (tr *TransformerJSONParse) jsonParseAll(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -167,6 +167,7 @@ func (tr *TransformerJSONParse) jsonParseAll(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerJSONParse) jsonParseSome(
@ -174,7 +175,7 @@ func (tr *TransformerJSONParse) jsonParseSome(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -190,4 +191,5 @@ func (tr *TransformerJSONParse) jsonParseSome(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

10
pkg/transformers/json_stringify.go
View file

@ -158,9 +158,9 @@ func (tr *TransformerJSONStringify) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerJSONStringify) jsonStringifyAll(
@ -168,7 +168,7 @@ func (tr *TransformerJSONStringify) jsonStringifyAll(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -178,6 +178,7 @@ func (tr *TransformerJSONStringify) jsonStringifyAll(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerJSONStringify) jsonStringifySome(
@ -185,7 +186,7 @@ func (tr *TransformerJSONStringify) jsonStringifySome(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -197,4 +198,5 @@ func (tr *TransformerJSONStringify) jsonStringifySome(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

3
pkg/transformers/label.go
View file

@ -115,11 +115,12 @@ func (tr *TransformerLabel) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
inrec.Label(tr.newNames)
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // including end-of-stream marker
return nil
}

3
pkg/transformers/latin1_to_utf8.go
View file

@ -88,7 +88,7 @@ func (tr *TransformerLatin1ToUTF8) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -110,4 +110,5 @@ func (tr *TransformerLatin1ToUTF8) Transform(
} else { // end of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

19
pkg/transformers/merge_fields.go
View file

@ -317,9 +317,9 @@ func (tr *TransformerMergeFields) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerMergeFields) transformByNameList(
@ -327,10 +327,10 @@ func (tr *TransformerMergeFields) transformByNameList(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if inrecAndContext.EndOfStream {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
return
return nil
}
inrec := inrecAndContext.Record
@ -370,6 +370,7 @@ func (tr *TransformerMergeFields) transformByNameList(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
func (tr *TransformerMergeFields) transformByNameRegex(
@ -377,10 +378,10 @@ func (tr *TransformerMergeFields) transformByNameRegex(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if inrecAndContext.EndOfStream {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
return
return nil
}
inrec := inrecAndContext.Record
@ -444,6 +445,7 @@ func (tr *TransformerMergeFields) transformByNameRegex(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
// mlr merge-fields -c in_,out_ -a sum
@ -457,10 +459,10 @@ func (tr *TransformerMergeFields) transformByCollapsing(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if inrecAndContext.EndOfStream {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
return
return nil
}
inrec := inrecAndContext.Record
@ -549,4 +551,5 @@ func (tr *TransformerMergeFields) transformByCollapsing(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}

5
pkg/transformers/most_or_least_frequent.go
View file

@ -217,13 +217,13 @@ func (tr *TransformerMostOrLeastFrequent) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok {
return
return nil
}
iCount, ok := tr.countsByGroup.GetWithCheck(groupingKey)
@ -291,4 +291,5 @@ func (tr *TransformerMostOrLeastFrequent) Transform(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // End-of-stream marker
}
return nil
}

36
pkg/transformers/nest.go
View file

@ -365,9 +365,9 @@ func (tr *TransformerNest) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
// getMatchingFieldNames returns field names matching tr.fieldRegex in record order.
@ -393,14 +393,14 @@ func (tr *TransformerNest) explodeValuesAcrossFields(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
fieldNames := tr.getMatchingFieldNames(inrec)
if len(fieldNames) == 0 {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
for _, fieldName := range fieldNames {
@ -430,6 +430,7 @@ func (tr *TransformerNest) explodeValuesAcrossFields(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerNest) explodeValuesAcrossRecords(
@ -437,20 +438,20 @@ func (tr *TransformerNest) explodeValuesAcrossRecords(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
fieldNames := tr.getMatchingFieldNames(inrec)
if len(fieldNames) == 0 {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
fieldName := fieldNames[0]
mvalue := inrec.Get(fieldName)
if mvalue == nil {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
svalue := mvalue.String()
@ -465,6 +466,7 @@ func (tr *TransformerNest) explodeValuesAcrossRecords(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerNest) explodePairsAcrossFields(
@ -472,14 +474,14 @@ func (tr *TransformerNest) explodePairsAcrossFields(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
fieldNames := tr.getMatchingFieldNames(inrec)
if len(fieldNames) == 0 {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
for _, fieldName := range fieldNames {
@ -517,6 +519,7 @@ func (tr *TransformerNest) explodePairsAcrossFields(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerNest) explodePairsAcrossRecords(
@ -524,20 +527,20 @@ func (tr *TransformerNest) explodePairsAcrossRecords(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
fieldNames := tr.getMatchingFieldNames(inrec)
if len(fieldNames) == 0 {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
fieldName := fieldNames[0]
mvalue := inrec.Get(fieldName)
if mvalue == nil {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
svalue := mvalue.String()
@ -563,6 +566,7 @@ func (tr *TransformerNest) explodePairsAcrossRecords(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerNest) implodeValuesAcrossFields(
@ -570,7 +574,7 @@ func (tr *TransformerNest) implodeValuesAcrossFields(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -611,6 +615,7 @@ func (tr *TransformerNest) implodeValuesAcrossFields(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerNest) implodeValueAcrossRecords(
@ -618,14 +623,14 @@ func (tr *TransformerNest) implodeValueAcrossRecords(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
originalEntry := inrec.GetEntry(tr.fieldName)
if originalEntry == nil {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
fieldValueCopy := originalEntry.Value.Copy()
@ -679,6 +684,7 @@ func (tr *TransformerNest) implodeValueAcrossRecords(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
type tNestBucket struct {

3
pkg/transformers/nothing.go
View file

@ -86,9 +86,10 @@ func (tr *TransformerNothing) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if inrecAndContext.EndOfStream {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

20
pkg/transformers/put_or_filter.go
View file

@ -531,7 +531,7 @@ func (tr *TransformerPut) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.runtimeState.OutputRecordsAndContexts = outputRecordsAndContexts
@ -545,8 +545,7 @@ func (tr *TransformerPut) Transform(
tr.runtimeState.Update(nil, &context)
err := tr.cstRootNode.ExecuteBeginBlocks(tr.runtimeState)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
tr.executedBeginBlocks = true
}
@ -556,8 +555,7 @@ func (tr *TransformerPut) Transform(
// Execute the main block on the current input record
outrec, err := tr.cstRootNode.ExecuteMainBlock(tr.runtimeState)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
if !tr.suppressOutputRecord {
@ -577,12 +575,11 @@ func (tr *TransformerPut) Transform(
if tr.runtimeState.FilterExpression.IsAbsent() {
filterBool = false
} else {
fmt.Fprintf(os.Stderr,
"Filter expression did not evaluate to boolean: got %s value %s",
return fmt.Errorf(
"mlr: filter expression did not evaluate to boolean: got %s value %s",
tr.runtimeState.FilterExpression.String(),
tr.runtimeState.FilterExpression.GetTypeName(),
)
os.Exit(1)
}
}
} else {
@ -606,16 +603,14 @@ func (tr *TransformerPut) Transform(
if !tr.executedBeginBlocks {
err := tr.cstRootNode.ExecuteBeginBlocks(tr.runtimeState)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
}
// Execute the end { ... } after the last input record
err := tr.cstRootNode.ExecuteEndBlocks(tr.runtimeState)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
// Send all registered OutputHandlerManager instances the end-of-stream
@ -624,4 +619,5 @@ func (tr *TransformerPut) Transform(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewEndOfStreamMarker(&context))
}
return nil
}

3
pkg/transformers/rank.go
View file

@ -170,13 +170,14 @@ func (tr *TransformerRank) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if tr.doSorted {
tr.transformSorted(inrecAndContext, outputRecordsAndContexts)
} else {
tr.transformUnsorted(inrecAndContext, outputRecordsAndContexts)
}
return nil
}
// transformSorted computes rank in a single pass, O(1) space, by comparing

3
pkg/transformers/regularize.go
View file

@ -91,7 +91,7 @@ func (tr *TransformerRegularize) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -113,4 +113,5 @@ func (tr *TransformerRegularize) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

3
pkg/transformers/remove_empty_columns.go
View file

@ -91,7 +91,7 @@ func (tr *TransformerRemoveEmptyColumns) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -123,4 +123,5 @@ func (tr *TransformerRemoveEmptyColumns) Transform(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // Emit the stream-terminating null record
}
return nil
}

10
pkg/transformers/rename.go
View file

@ -189,9 +189,9 @@ func (tr *TransformerRename) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerRename) transformWithoutRegexes(
@ -199,7 +199,7 @@ func (tr *TransformerRename) transformWithoutRegexes(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -212,6 +212,7 @@ func (tr *TransformerRename) transformWithoutRegexes(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // including end-of-stream marker
return nil
}
func (tr *TransformerRename) transformWithRegexes(
@ -219,7 +220,7 @@ func (tr *TransformerRename) transformWithRegexes(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -248,4 +249,5 @@ func (tr *TransformerRename) transformWithRegexes(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // including end-of-stream marker
}
return nil
}

3
pkg/transformers/reorder.go
View file

@ -210,7 +210,7 @@ func (tr *TransformerReorder) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.recordTransformerFunc(
@ -220,6 +220,7 @@ func (tr *TransformerReorder) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerReorder) reorderToStartNoRegex(

14
pkg/transformers/repeat.go
View file

@ -173,9 +173,9 @@ func (tr *TransformerRepeat) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerRepeat) repeatByCount(
@ -183,7 +183,7 @@ func (tr *TransformerRepeat) repeatByCount(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
for i := int64(0); i < tr.repeatCount; i++ {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(
@ -194,6 +194,7 @@ func (tr *TransformerRepeat) repeatByCount(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerRepeat) repeatByFieldName(
@ -201,15 +202,15 @@ func (tr *TransformerRepeat) repeatByFieldName(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
fieldValue := inrecAndContext.Record.Get(tr.repeatCountFieldName)
if fieldValue == nil {
return
return nil
}
repeatCount, ok := fieldValue.GetIntValue()
if !ok {
return
return nil
}
for i := 0; i < int(repeatCount); i++ {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(
@ -221,4 +222,5 @@ func (tr *TransformerRepeat) repeatByFieldName(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

15
pkg/transformers/reshape.go
View file

@ -300,9 +300,9 @@ func (tr *TransformerReshape) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerReshape) wideToLongNoRegex(
@ -310,7 +310,7 @@ func (tr *TransformerReshape) wideToLongNoRegex(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
pairs := mlrval.NewMlrmap()
@ -341,6 +341,7 @@ func (tr *TransformerReshape) wideToLongNoRegex(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerReshape) wideToLongRegex(
@ -348,7 +349,7 @@ func (tr *TransformerReshape) wideToLongRegex(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
pairs := mlrval.NewMlrmap()
@ -382,6 +383,7 @@ func (tr *TransformerReshape) wideToLongRegex(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
func (tr *TransformerReshape) longToWide(
@ -389,7 +391,7 @@ func (tr *TransformerReshape) longToWide(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -397,7 +399,7 @@ func (tr *TransformerReshape) longToWide(
splitOutValueFieldValue := inrec.Get(tr.splitOutValueFieldName)
if splitOutKeyFieldValue == nil || splitOutValueFieldValue == nil {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
inrec.Remove(tr.splitOutKeyFieldName)
@ -443,6 +445,7 @@ func (tr *TransformerReshape) longToWide(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // emit end-of-stream marker
}
return nil
}
type tReshapeBucket struct {

3
pkg/transformers/sample.go
View file

@ -144,7 +144,7 @@ func (tr *TransformerSample) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
// Not end of input stream: retain the record, and emit nothing until end of stream.
if !inrecAndContext.EndOfStream {
@ -172,6 +172,7 @@ func (tr *TransformerSample) Transform(
// Emit the stream-terminating null record
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func newSampleBucket(sampleCount int64) *sampleBucketType {

3
pkg/transformers/sec2gmt.go
View file

@ -156,7 +156,7 @@ func (tr *TransformerSec2GMT) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -178,4 +178,5 @@ func (tr *TransformerSec2GMT) Transform(
} else { // End of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

3
pkg/transformers/sec2gmtdate.go
View file

@ -104,7 +104,7 @@ func (tr *TransformerSec2GMTDate) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -119,4 +119,5 @@ func (tr *TransformerSec2GMTDate) Transform(
} else { // End of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

5
pkg/transformers/seqgen.go
View file

@ -185,10 +185,10 @@ func (tr *TransformerSeqgen) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
// Discard upstream records; generate output only when upstream is done.
return
return nil
}
counter := tr.start
@ -231,4 +231,5 @@ func (tr *TransformerSeqgen) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewEndOfStreamMarker(context))
return nil
}

3
pkg/transformers/shuffle.go
View file

@ -94,7 +94,7 @@ func (tr *TransformerShuffle) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
// Not end of input stream: retain the record, and emit nothing until end of stream.
if !inrecAndContext.EndOfStream {
@ -136,4 +136,5 @@ func (tr *TransformerShuffle) Transform(
// Emit the stream-terminating null record
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

3
pkg/transformers/skip_trivial_records.go
View file

@ -92,7 +92,7 @@ func (tr *TransformerSkipTrivialRecords) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -111,4 +111,5 @@ func (tr *TransformerSkipTrivialRecords) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

5
pkg/transformers/sort.go
View file

@ -394,7 +394,7 @@ func (tr *TransformerSort) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -411,7 +411,7 @@ func (tr *TransformerSort) Transform(
)
if !ok {
tr.spillGroup = append(tr.spillGroup, inrecAndContext)
return
return nil
}
recordListForGroup := tr.recordListsByGroup.Get(groupingKey)
@ -468,6 +468,7 @@ func (tr *TransformerSort) Transform(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func groupHeadsToArray(groupHeads *lib.OrderedMap[[]*mlrval.Mlrval]) []GroupingKeysAndMlrvals {

16
pkg/transformers/sort_within_records.go
View file

@ -188,9 +188,9 @@ func (tr *TransformerSortWithinRecords) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
// ----------------------------------------------------------------
@ -243,7 +243,7 @@ func (tr *TransformerSortWithinRecords) transformSelective(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
var matchingKeys []string
@ -266,6 +266,7 @@ func (tr *TransformerSortWithinRecords) transformSelective(
*inrec = *other
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
// ----------------------------------------------------------------
@ -274,7 +275,7 @@ func (tr *TransformerSortWithinRecords) transformSelectiveRecursively(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
var matchingKeys []string
@ -308,6 +309,7 @@ func (tr *TransformerSortWithinRecords) transformSelectiveRecursively(
*inrec = *other
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
// ----------------------------------------------------------------
@ -316,7 +318,7 @@ func (tr *TransformerSortWithinRecords) transformNonrecursively(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
if tr.doNatural {
@ -326,6 +328,7 @@ func (tr *TransformerSortWithinRecords) transformNonrecursively(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // including end-of-stream marker
return nil
}
func (tr *TransformerSortWithinRecords) transformRecursively(
@ -333,7 +336,7 @@ func (tr *TransformerSortWithinRecords) transformRecursively(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
if tr.doNatural {
@ -343,4 +346,5 @@ func (tr *TransformerSortWithinRecords) transformRecursively(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // including end-of-stream marker
return nil
}

5
pkg/transformers/sparkline.go
View file

@ -119,7 +119,7 @@ func (tr *TransformerSparkline) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
@ -130,7 +130,7 @@ func (tr *TransformerSparkline) Transform(
tr.valuesByField[fieldName] = append(tr.valuesByField[fieldName], mvalue.Copy())
}
}
return
return nil
}
// Else, end of stream: emit one summary record per field.
@ -155,6 +155,7 @@ func (tr *TransformerSparkline) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // Emit the end-of-stream marker
return nil
}
// floatRangeOf returns the min and max of the numeric values, and whether

11
pkg/transformers/sparsify.go
View file

@ -136,11 +136,11 @@ func (tr *TransformerSparsify) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.recordTransformerFunc(
return tr.recordTransformerFunc(
inrecAndContext,
outputRecordsAndContexts,
inputDownstreamDoneChannel,
@ -149,6 +149,7 @@ func (tr *TransformerSparsify) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerSparsify) transformAll(
@ -156,7 +157,7 @@ func (tr *TransformerSparsify) transformAll(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
outrec := mlrval.NewMlrmapAsRecord()
@ -169,6 +170,7 @@ func (tr *TransformerSparsify) transformAll(
outrecAndContext := types.NewRecordAndContext(outrec, &inrecAndContext.Context)
*outputRecordsAndContexts = append(*outputRecordsAndContexts, outrecAndContext)
return nil
}
func (tr *TransformerSparsify) transformSome(
@ -176,7 +178,7 @@ func (tr *TransformerSparsify) transformSome(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
outrec := mlrval.NewMlrmapAsRecord()
@ -193,4 +195,5 @@ func (tr *TransformerSparsify) transformSome(
outrecAndContext := types.NewRecordAndContext(outrec, &inrecAndContext.Context)
*outputRecordsAndContexts = append(*outputRecordsAndContexts, outrecAndContext)
return nil
}

41
pkg/transformers/split.go
View file

@ -320,9 +320,9 @@ func (tr *TransformerSplit) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel,
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel,
outputDownstreamDoneChannel)
}
@ -331,7 +331,7 @@ func (tr *TransformerSplit) splitModUngrouped(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
remainder := 1 + (tr.ungroupedCounter % tr.n)
filename := tr.makeUngroupedOutputFileName(remainder)
@ -345,8 +345,7 @@ func (tr *TransformerSplit) splitModUngrouped(
}
err := tr.outputHandlerManager.WriteRecordAndContext(recordAndContextForWriter, filename)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
if tr.emitDownstream {
@ -359,12 +358,14 @@ func (tr *TransformerSplit) splitModUngrouped(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
errs := tr.outputHandlerManager.Close()
if len(errs) > 0 {
for _, err := range errs {
// Print any additional errors here; return the first one.
for _, err := range errs[1:] {
fmt.Fprintf(os.Stderr, "mlr: file-close error: %v\n", err)
}
os.Exit(1)
return fmt.Errorf("mlr: file-close error: %v", errs[0])
}
}
return nil
}
func (tr *TransformerSplit) splitSizeUngrouped(
@ -372,7 +373,7 @@ func (tr *TransformerSplit) splitSizeUngrouped(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
var err error
if !inrecAndContext.EndOfStream {
quotient := 1 + (tr.ungroupedCounter / tr.n)
@ -381,8 +382,7 @@ func (tr *TransformerSplit) splitSizeUngrouped(
if tr.outputHandler != nil {
err = tr.outputHandler.Close()
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
}
@ -393,8 +393,7 @@ func (tr *TransformerSplit) splitSizeUngrouped(
tr.doAppend,
)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
tr.previousQuotient = quotient
@ -409,8 +408,7 @@ func (tr *TransformerSplit) splitSizeUngrouped(
}
err = tr.outputHandler.WriteRecordAndContext(recordAndContextForWriter)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
if tr.emitDownstream {
@ -425,11 +423,11 @@ func (tr *TransformerSplit) splitSizeUngrouped(
if tr.outputHandler != nil {
err := tr.outputHandler.Close()
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
}
}
return nil
}
func (tr *TransformerSplit) splitGrouped(
@ -437,7 +435,7 @@ func (tr *TransformerSplit) splitGrouped(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
var filename string
groupByFieldValues, ok := inrecAndContext.Record.GetSelectedValues(tr.groupByFieldNames)
@ -460,8 +458,7 @@ func (tr *TransformerSplit) splitGrouped(
}
err := tr.outputHandlerManager.WriteRecordAndContext(recordAndContextForWriter, filename)
if err != nil {
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return err
}
if tr.emitDownstream {
@ -473,12 +470,14 @@ func (tr *TransformerSplit) splitGrouped(
errs := tr.outputHandlerManager.Close()
if len(errs) > 0 {
for _, err := range errs {
// Print any additional errors here; return the first one.
for _, err := range errs[1:] {
fmt.Fprintf(os.Stderr, "mlr: file-close error: %v\n", err)
}
os.Exit(1)
return fmt.Errorf("mlr: file-close error: %v", errs[0])
}
}
return nil
}
// makeUngroupedOutputFileName example: "split_53.csv" or "folder/split_53.csv" with --folder

3
pkg/transformers/stats1.go
View file

@ -399,13 +399,14 @@ func (tr *TransformerStats1) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.handleInputRecord(inrecAndContext, outputRecordsAndContexts)
} else {
tr.handleEndOfRecordStream(inrecAndContext, outputRecordsAndContexts)
}
return nil
}
func (tr *TransformerStats1) handleInputRecord(

4
pkg/transformers/stats1_test.go
View file

@ -26,12 +26,12 @@ func runStats1TestTransformer(
outputDownstreamDoneChannel := make(chan bool, 1)
outputs := make([]*types.RecordAndContext, 0)
for _, input := range inputs {
tr.Transform(input, &outputs, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
_ = tr.Transform(input, &outputs, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
// End-of-stream marker
eos := types.NewRecordAndContext(nil, types.NewNilContext())
eos.EndOfStream = true
tr.Transform(eos, &outputs, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
_ = tr.Transform(eos, &outputs, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return outputs
}

15
pkg/transformers/stats2.go
View file

@ -261,11 +261,13 @@ func (tr *TransformerStats2) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.ingest(inrecAndContext)
if err := tr.ingest(inrecAndContext); err != nil {
return err
}
if tr.doIterativeStats {
// The input record is modified in this case, with new fields appended
@ -283,18 +285,19 @@ func (tr *TransformerStats2) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerStats2) ingest(
inrecAndContext *types.RecordAndContext,
) {
) error {
inrec := inrecAndContext.Record
// E.g. if grouping by "a" and "b", and the current record has a=circle, b=blue,
// then groupingKey is the string "circle,blue".
groupingKey, groupByFieldValues, ok := inrec.GetSelectedValuesAndJoined(tr.groupByFieldNameList)
if !ok {
return
return nil
}
tr.groupingKeysToGroupByFieldValues.Put(groupingKey, groupByFieldValues)
@ -349,8 +352,7 @@ func (tr *TransformerStats2) ingest(
tr.doVerbose,
)
if accumulator == nil {
fmt.Fprintf(os.Stderr, "mlr stats2: accumulator creation failed\n")
os.Exit(1)
return cli.VerbErrorf(verbNameStats2, "accumulator creation failed")
}
valueFieldsToAccumulator.Put(accumulatorName, accumulator)
}
@ -369,6 +371,7 @@ func (tr *TransformerStats2) ingest(
)
}
}
return nil
}
func (tr *TransformerStats2) emit(

105
pkg/transformers/step.go
View file

@ -184,6 +184,11 @@ func transformerStepParseCLI(
verb, "stepper \"%s\": count must be a positive integer", stepperName,
)
}
if stepperNameHasNegativeSlwin(stepperName) {
return nil, cli.VerbErrorf(
verb, "stepper needed non-negative num-backward & num-forward in %s.", stepperName,
)
}
return nil, cli.VerbErrorf(verb, "stepper \"%s\" not found", stepperName)
}
stepperInputs = append(stepperInputs, stepperInput)
@ -366,11 +371,13 @@ func (tr *TransformerStep) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.handleRecord(inrecAndContext, outputRecordsAndContexts)
if err := tr.handleRecord(inrecAndContext, outputRecordsAndContexts); err != nil {
return err
}
} else {
// As described in comments at the top of this file: process through all delayed-input
@ -384,8 +391,9 @@ func (tr *TransformerStep) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
return nil
}
// handleRecord processes records received before the end of the record stream is seen.
@ -395,7 +403,7 @@ func (tr *TransformerStep) Transform(
func (tr *TransformerStep) handleRecord(
inrecAndContext *types.RecordAndContext,
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
) {
) error {
inrec := inrecAndContext.Record
// Group-by field names are ["a", "b"]
@ -404,7 +412,7 @@ func (tr *TransformerStep) handleRecord(
groupingKey, gok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !gok { // current record doesn't have fields to be stepped; pass it along
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
// Create the data structure on first reference
@ -459,15 +467,18 @@ func (tr *TransformerStep) handleRecord(
for _, stepperInput := range tr.stepperInputs {
stepper, present := accFieldToAccState[stepperInput.name]
if !present {
stepper = allocateStepper(
var err error
stepper, err = allocateStepper(
stepperInput,
valueFieldName,
tr.stringAlphas,
tr.ewmaSuffixes,
)
if err != nil {
return err
}
if stepper == nil {
fmt.Fprintf(os.Stderr, "mlr step: stepper allocation failed\n")
os.Exit(1)
return cli.VerbErrorf(verbNameStep, "stepper allocation failed")
}
accFieldToAccState[stepperInput.name] = stepper
}
@ -481,6 +492,7 @@ func (tr *TransformerStep) handleRecord(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, outrecAndContext)
tr.removeFromLog(outrecAndContext)
}
return nil
}
// handleDrainRecord processes records received after the end of the record stream is seen. The
@ -584,7 +596,7 @@ type tStepperAllocator func(
inputFieldName string,
stringAlphas []string,
ewmaSuffixes []string,
) tStepper
) (tStepper, error)
type tStepperInput struct {
name string
@ -720,7 +732,7 @@ func allocateStepper(
inputFieldName string,
stringAlphas []string,
ewmaSuffixes []string,
) tStepper {
) (tStepper, error) {
for _, stepperLookup := range STEPPER_LOOKUP_TABLE {
if stepperLookup.nameIsVariable {
if stepperLookup.ownsPrefix(stepperInput.name) {
@ -742,7 +754,7 @@ func allocateStepper(
}
}
}
return nil
return nil, nil
}
// parseStepperCount parses stepper names which accept an optional trailing
@ -870,13 +882,13 @@ func stepperDeltaAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
count, _ := parseStepperCount(stepperInput.name, "delta")
return &tStepperDelta{
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_" + stepperInput.name,
prevValues: newValueRing(count),
}
}, nil
}
func (stepper *tStepperDelta) process(
@ -976,13 +988,13 @@ func stepperShiftAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
count, _ := parseStepperCount(stepperInput.name, "shift")
return &tStepperShiftLag{
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_" + stepperInput.name,
prevValues: newValueRing(count),
}
}, nil
}
func stepperShiftLagAlloc(
@ -990,13 +1002,13 @@ func stepperShiftLagAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
count, _ := parseStepperCount(stepperInput.name, "shift_lag")
return &tStepperShiftLag{
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_" + stepperInput.name,
prevValues: newValueRing(count),
}
}, nil
}
func (stepper *tStepperShiftLag) process(
@ -1062,12 +1074,12 @@ func stepperShiftLeadAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
return &tStepperShiftLead{
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_" + stepperInput.name,
numRecordsForward: stepperInput.numRecordsForward,
}
}, nil
}
func (stepper *tStepperShiftLead) process(
@ -1114,12 +1126,12 @@ func stepperFromFirstAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
return &tStepperFromFirst{
first: nil,
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_from_first",
}
}, nil
}
func (stepper *tStepperFromFirst) process(
@ -1185,13 +1197,13 @@ func stepperRatioAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
count, _ := parseStepperCount(stepperInput.name, "ratio")
return &tStepperRatio{
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_" + stepperInput.name,
prevValues: newValueRing(count),
}
}, nil
}
func (stepper *tStepperRatio) process(
@ -1252,12 +1264,12 @@ func stepperRprodAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
return &tStepperRprod{
rprod: mlrval.FromInt(1),
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_rprod",
}
}, nil
}
func (stepper *tStepperRprod) process(
@ -1307,12 +1319,12 @@ func stepperRsumAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
return &tStepperRsum{
rsum: mlrval.FromInt(0),
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_rsum",
}
}, nil
}
func (stepper *tStepperRsum) process(
@ -1362,12 +1374,12 @@ func stepperCounterAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
return &tStepperCounter{
counter: mlrval.FromInt(0),
inputFieldName: inputFieldName,
outputFieldName: inputFieldName + "_counter",
}
}, nil
}
func (stepper *tStepperCounter) process(
@ -1422,7 +1434,7 @@ func stepperEWMAAlloc(
inputFieldName string,
stringAlphas []string,
ewmaSuffixes []string,
) tStepper {
) (tStepper, error) {
// We trust our caller has already checked len(stringAlphas) == len(ewmaSuffixes) in the CLI
// parser.
@ -1443,12 +1455,8 @@ func stepperEWMAAlloc(
dalpha, ok := lib.TryFloatFromString(stringAlpha)
if !ok {
fmt.Fprintf(
os.Stderr,
"mlr step: could not parse \"%s\" as floating-point EWMA coefficient.\n",
stringAlpha,
)
os.Exit(1)
return nil, cli.VerbErrorf(verbNameStep,
"could not parse \"%s\" as floating-point EWMA coefficient.", stringAlpha)
}
alphas[i] = mlrval.FromFloat(dalpha)
oneMinusAlphas[i] = mlrval.FromFloat(1.0 - dalpha)
@ -1463,7 +1471,7 @@ func stepperEWMAAlloc(
inputFieldName: inputFieldName,
outputFieldNames: outputFieldNames,
havePrevs: false,
}
}, nil
}
func (stepper *tStepperEWMA) process(
@ -1516,6 +1524,15 @@ func stepperSlwintOwnsPrefix(
return strings.HasPrefix(stepperName, "slwin")
}
// stepperNameHasNegativeSlwin detects slwin stepper names whose
// num-backward/num-forward parsed but were negative, so the CLI parser can
// give a more specific error than "stepper not found".
func stepperNameHasNegativeSlwin(stepperName string) bool {
var numRecordsBackward, numRecordsForward int
n, err := fmt.Sscanf(stepperName, "slwin_%d_%d", &numRecordsBackward, &numRecordsForward)
return n == 2 && err == nil && (numRecordsBackward < 0 || numRecordsForward < 0)
}
func stepperSlwinInputFromName(
stepperName string,
) *tStepperInput {
@ -1523,13 +1540,9 @@ func stepperSlwinInputFromName(
n, err := fmt.Sscanf(stepperName, "slwin_%d_%d", &numRecordsBackward, &numRecordsForward)
if n == 2 && err == nil {
if numRecordsBackward < 0 || numRecordsForward < 0 {
fmt.Fprintf(
os.Stderr,
"mlr %s: stepper needed non-negative num-backward & num-forward in %s.\n",
verbNameStep,
stepperName,
)
os.Exit(1)
// The CLI parser reports the specific negative-parameter error;
// see stepperNameHasNegativeSlwin.
return nil
}
return &tStepperInput{
name: stepperName,
@ -1546,7 +1559,7 @@ func stepperSlwinAlloc(
inputFieldName string,
_unused1 []string,
_unused2 []string,
) tStepper {
) (tStepper, error) {
nb := stepperInput.numRecordsBackward
nf := stepperInput.numRecordsForward
return &tStepperSlwin{
@ -1554,7 +1567,7 @@ func stepperSlwinAlloc(
outputFieldName: fmt.Sprintf("%s_%d_%d", inputFieldName, nb, nf),
numRecordsBackward: nb,
numRecordsForward: nf,
}
}, nil
}
func (stepper *tStepperSlwin) process(

6
pkg/transformers/step_test.go
View file

@ -137,7 +137,11 @@ func TestAllocateStepperOutputFieldNames(t *testing.T) {
t.Errorf("stepperInputFromName(%q) = nil; want non-nil", tc.stepperName)
continue
}
stepper := allocateStepper(stepperInput, "x", nil, nil)
stepper, err := allocateStepper(stepperInput, "x", nil, nil)
if err != nil {
t.Errorf("allocateStepper for %q: %v", tc.stepperName, err)
continue
}
if stepper == nil {
t.Errorf("allocateStepper for %q = nil; want non-nil", tc.stepperName)
continue

3
pkg/transformers/subs.go
View file

@ -323,7 +323,7 @@ func (tr *TransformerSubs) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
@ -337,6 +337,7 @@ func (tr *TransformerSubs) Transform(
}
// Including emit of end-of-stream marker
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return nil
}
// fieldAcceptorByNames implements -f

3
pkg/transformers/summary.go
View file

@ -295,7 +295,7 @@ func (tr *TransformerSummary) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.ingest(inrecAndContext)
@ -306,6 +306,7 @@ func (tr *TransformerSummary) Transform(
tr.emit(inrecAndContext, outputRecordsAndContexts)
}
}
return nil
}
func (tr *TransformerSummary) ingest(

12
pkg/transformers/surv.go
View file

@ -112,20 +112,20 @@ func (tr *TransformerSurv) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext,
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
rec := inrecAndContext.Record
mvDur := rec.Get(tr.durationField)
if mvDur == nil {
// Skip records missing the duration field
return
return nil
}
duration := mvDur.GetNumericToFloatValueOrDie()
mvStat := rec.Get(tr.statusField)
if mvStat == nil {
// Skip records missing the status field
return
return nil
}
status := mvStat.GetNumericToFloatValueOrDie() != 0
tr.times = append(tr.times, duration)
@ -135,7 +135,7 @@ func (tr *TransformerSurv) Transform(
n := len(tr.times)
if n == 0 {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
return
return nil
}
durations := tr.times
statuses := make([]float64, n)
@ -151,8 +151,7 @@ func (tr *TransformerSurv) Transform(
ds := statmodel.NewDataset(dataCols, names)
sf, err := duration.NewSurvfuncRight(ds, tr.durationField, tr.statusField, &duration.SurvfuncRightConfig{})
if err != nil {
fmt.Fprintf(os.Stderr, "mlr surv: %v\n", err)
os.Exit(1)
return cli.VerbErrorf(verbNameSurv, "%v", err)
}
sf.Fit()
times := sf.Time()
@ -165,4 +164,5 @@ func (tr *TransformerSurv) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

3
pkg/transformers/tac.go
View file

@ -87,7 +87,7 @@ func (tr *TransformerTac) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.recordsAndContexts = append(tr.recordsAndContexts, inrecAndContext)
@ -98,4 +98,5 @@ func (tr *TransformerTac) Transform(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewEndOfStreamMarker(&inrecAndContext.Context))
}
return nil
}

14
pkg/transformers/tail.go
View file

@ -154,9 +154,9 @@ func (tr *TransformerTail) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerTail) transformLastN(
@ -164,13 +164,13 @@ func (tr *TransformerTail) transformLastN(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok {
return
return nil
}
recordListForGroup := tr.recordListsByGroup.Get(groupingKey)
@ -192,6 +192,7 @@ func (tr *TransformerTail) transformLastN(
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerTail) transformFromStart(
@ -199,13 +200,13 @@ func (tr *TransformerTail) transformFromStart(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, ok := inrec.GetSelectedValuesJoined(tr.groupByFieldNames)
if !ok {
return
return nil
}
tr.countsByGroup[groupingKey]++
@ -217,4 +218,5 @@ func (tr *TransformerTail) transformFromStart(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

19
pkg/transformers/tee.go
View file

@ -172,7 +172,7 @@ func (tr *TransformerTee) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
// If we receive a downstream-done flag from a transformer downstream from
// us, read it to unblock their goroutine but -- unlike most other verbs --
@ -202,25 +202,22 @@ func (tr *TransformerTee) Transform(
// #1671).
err := tr.fileOutputHandler.WriteRecordAndContext(inrecAndContext.Copy())
if err != nil {
fmt.Fprintf(
os.Stderr,
"%s: error writing to tee \"%s\":\n",
"mlr", tr.filenameOrCommandForDisplay,
return fmt.Errorf(
"%s: error writing to tee \"%s\":\n%v",
"mlr", tr.filenameOrCommandForDisplay, err,
)
os.Exit(1)
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
} else {
err := tr.fileOutputHandler.Close()
if err != nil {
fmt.Fprintf(
os.Stderr,
"%s: error closing tee \"%s\":\n",
"mlr", tr.filenameOrCommandForDisplay,
return fmt.Errorf(
"%s: error closing tee \"%s\":\n%v",
"mlr", tr.filenameOrCommandForDisplay, err,
)
os.Exit(1)
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

3
pkg/transformers/template.go
View file

@ -143,7 +143,7 @@ func (tr *TransformerTemplate) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -161,4 +161,5 @@ func (tr *TransformerTemplate) Transform(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

3
pkg/transformers/top.go
View file

@ -186,13 +186,14 @@ func (tr *TransformerTop) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.ingest(inrecAndContext)
} else {
tr.emit(inrecAndContext, outputRecordsAndContexts)
}
return nil
}
func (tr *TransformerTop) ingest(

10
pkg/transformers/unflatten.go
View file

@ -140,9 +140,9 @@ func (tr *TransformerUnflatten) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerUnflatten) unflattenAll(
@ -150,7 +150,7 @@ func (tr *TransformerUnflatten) unflattenAll(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
oFlatSep := tr.oFlatSep
@ -162,6 +162,7 @@ func (tr *TransformerUnflatten) unflattenAll(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerUnflatten) unflattenSome(
@ -169,7 +170,7 @@ func (tr *TransformerUnflatten) unflattenSome(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
oFlatSep := tr.oFlatSep
@ -181,4 +182,5 @@ func (tr *TransformerUnflatten) unflattenSome(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

27
pkg/transformers/uniq.go
View file

@ -382,9 +382,9 @@ func (tr *TransformerUniq) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
// Print each unique record only once, with uniqueness counts. This means
@ -394,7 +394,7 @@ func (tr *TransformerUniq) transformUniqifyEntireRecordsShowCounts(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -420,6 +420,7 @@ func (tr *TransformerUniq) transformUniqifyEntireRecordsShowCounts(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
// Print count of unique records. This means non-streaming, with output at end
@ -429,7 +430,7 @@ func (tr *TransformerUniq) transformUniqifyEntireRecordsShowNumDistinctOnly(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
recordAsString := inrec.String()
@ -447,6 +448,7 @@ func (tr *TransformerUniq) transformUniqifyEntireRecordsShowNumDistinctOnly(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
// Print each unique record only once (on first occurrence).
@ -455,7 +457,7 @@ func (tr *TransformerUniq) transformUniqifyEntireRecords(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -469,6 +471,7 @@ func (tr *TransformerUniq) transformUniqifyEntireRecords(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerUniq) transformUnlashed(
@ -476,7 +479,7 @@ func (tr *TransformerUniq) transformUnlashed(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -523,6 +526,7 @@ func (tr *TransformerUniq) transformUnlashed(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerUniq) transformNumDistinctOnly(
@ -530,7 +534,7 @@ func (tr *TransformerUniq) transformNumDistinctOnly(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -554,6 +558,7 @@ func (tr *TransformerUniq) transformNumDistinctOnly(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerUniq) transformWithCounts(
@ -561,7 +566,7 @@ func (tr *TransformerUniq) transformWithCounts(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -603,6 +608,7 @@ func (tr *TransformerUniq) transformWithCounts(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerUniq) transformWithoutCounts(
@ -610,13 +616,13 @@ func (tr *TransformerUniq) transformWithoutCounts(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
groupingKey, selectedValues, ok := inrec.GetSelectedValuesAndJoined(tr.getFieldNamesForGrouping(inrec))
if !ok {
return
return nil
}
iCount, present := tr.countsByGroup.GetWithCheck(groupingKey)
@ -641,4 +647,5 @@ func (tr *TransformerUniq) transformWithoutCounts(
} else { // end of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

14
pkg/transformers/unspace.go
View file

@ -122,10 +122,10 @@ func (tr *TransformerUnspace) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
tr.recordTransformerFunc(
return tr.recordTransformerFunc(
inrecAndContext,
outputRecordsAndContexts,
inputDownstreamDoneChannel,
@ -134,6 +134,7 @@ func (tr *TransformerUnspace) Transform(
} else { // end of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}
func (tr *TransformerUnspace) transformKeysOnly(
@ -141,7 +142,7 @@ func (tr *TransformerUnspace) transformKeysOnly(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
_ <-chan bool,
__ chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
newrec := mlrval.NewMlrmapAsRecord()
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -150,6 +151,7 @@ func (tr *TransformerUnspace) transformKeysOnly(
newrec.PutReference(newkey, pe.Value)
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(newrec, &inrecAndContext.Context))
return nil
}
func (tr *TransformerUnspace) transformValuesOnly(
@ -157,7 +159,7 @@ func (tr *TransformerUnspace) transformValuesOnly(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
_ <-chan bool,
__ chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
stringval, ok := pe.Value.GetStringValue()
@ -166,6 +168,7 @@ func (tr *TransformerUnspace) transformValuesOnly(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(inrec, &inrecAndContext.Context))
return nil
}
func (tr *TransformerUnspace) transformKeysAndValues(
@ -173,7 +176,7 @@ func (tr *TransformerUnspace) transformKeysAndValues(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
_ <-chan bool,
__ chan<- bool,
) {
) error {
inrec := inrecAndContext.Record
newrec := mlrval.NewMlrmapAsRecord()
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -187,6 +190,7 @@ func (tr *TransformerUnspace) transformKeysAndValues(
}
}
*outputRecordsAndContexts = append(*outputRecordsAndContexts, types.NewRecordAndContext(newrec, &inrecAndContext.Context))
return nil
}
func (tr *TransformerUnspace) unspace(input string) string {

10
pkg/transformers/unsparsify.go
View file

@ -147,9 +147,9 @@ func (tr *TransformerUnsparsify) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
return tr.recordTransformerFunc(inrecAndContext, outputRecordsAndContexts, inputDownstreamDoneChannel, outputDownstreamDoneChannel)
}
func (tr *TransformerUnsparsify) transformNonStreaming(
@ -157,7 +157,7 @@ func (tr *TransformerUnsparsify) transformNonStreaming(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
for pe := inrec.Head; pe != nil; pe = pe.Next {
@ -186,6 +186,7 @@ func (tr *TransformerUnsparsify) transformNonStreaming(
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}
func (tr *TransformerUnsparsify) transformStreaming(
@ -193,7 +194,7 @@ func (tr *TransformerUnsparsify) transformStreaming(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -208,4 +209,5 @@ func (tr *TransformerUnsparsify) transformStreaming(
} else {
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext) // end-of-stream marker
}
return nil
}

3
pkg/transformers/utf8_to_latin1.go
View file

@ -88,7 +88,7 @@ func (tr *TransformerUTF8ToLatin1) Transform(
outputRecordsAndContexts *[]*types.RecordAndContext, // list of *types.RecordAndContext
inputDownstreamDoneChannel <-chan bool,
outputDownstreamDoneChannel chan<- bool,
) {
) error {
HandleDefaultDownstreamDone(inputDownstreamDoneChannel, outputDownstreamDoneChannel)
if !inrecAndContext.EndOfStream {
inrec := inrecAndContext.Record
@ -110,4 +110,5 @@ func (tr *TransformerUTF8ToLatin1) Transform(
} else { // end of record stream
*outputRecordsAndContexts = append(*outputRecordsAndContexts, inrecAndContext)
}
return nil
}

110
pkg/transformers/utils/join_bucket_keeper.go
View file

@ -105,8 +105,6 @@
package utils
import (
"fmt"
"os"
"strings"
"github.com/johnkerl/miller/v6/pkg/cli"
@ -159,13 +157,12 @@ func NewJoinBucketKeeper(
joinReaderOptions *cli.TReaderOptions,
leftJoinFieldNames []string,
leftKeepFieldNameSet map[string]bool,
) *JoinBucketKeeper {
) (*JoinBucketKeeper, error) {
// Instantiate the record-reader
recordReader, err := input.Create(joinReaderOptions, 1) // TODO: maybe increase records per batch
if err != nil {
fmt.Fprintf(os.Stderr, "mlr join: %v\n", err)
os.Exit(1)
return nil, cli.VerbErrorf("join", "%v", err)
}
// Set the initial context for the left-file. Since Go is concurrent, the
@ -201,7 +198,7 @@ func NewJoinBucketKeeper(
state: LEFT_STATE_0_PREFILL,
}
return keeper
return keeper, nil
}
// For JoinBucketKeeper state machine
@ -246,7 +243,7 @@ func (keeper *JoinBucketKeeper) computeState() tJoinBucketKeeperState {
func (keeper *JoinBucketKeeper) FindJoinBucket(
rightFieldValues []*mlrval.Mlrval, // nil means right-file EOF
) bool {
) (bool, error) {
// TODO: comment me
isPaired := false
@ -254,9 +251,13 @@ func (keeper *JoinBucketKeeper) FindJoinBucket(
// to be had) but it may or may not make the join keys from the current
// right record.
if keeper.state == LEFT_STATE_0_PREFILL {
keeper.prepareForFirstJoinBucket()
if err := keeper.prepareForFirstJoinBucket(); err != nil {
return false, err
}
if keeper.peekRecordAndContext != nil {
keeper.fillNextJoinBucket()
if err := keeper.fillNextJoinBucket(); err != nil {
return false, err
}
}
keeper.state = keeper.computeState()
}
@ -272,10 +273,14 @@ func (keeper *JoinBucketKeeper) FindJoinBucket(
// Example: joining on "id" column and left file has several
// join-field records with id=3, then several with id=7, but
// the current right record has id=5.
keeper.prepareForNewJoinBucket(rightFieldValues)
if err := keeper.prepareForNewJoinBucket(rightFieldValues); err != nil {
return false, err
}
if keeper.peekRecordAndContext != nil {
keeper.fillNextJoinBucket()
if err := keeper.fillNextJoinBucket(); err != nil {
return false, err
}
}
// TODO: privatize more
@ -300,33 +305,35 @@ func (keeper *JoinBucketKeeper) FindJoinBucket(
// and no need to advance left.
isPaired = false
}
} else if keeper.state != LEFT_STATE_3_EOF {
fmt.Fprintf(
os.Stderr,
"%s: internal coding error: failed transition from prefill state.\n",
"mlr",
)
os.Exit(1)
} else {
lib.InternalCodingErrorWithMessageIf(keeper.state != LEFT_STATE_3_EOF,
"failed transition from prefill state")
}
} else { // Right EOF
keeper.markRemainingsAsUnpaired()
if err := keeper.markRemainingsAsUnpaired(); err != nil {
return false, err
}
}
keeper.state = keeper.computeState()
return isPaired
return isPaired, nil
}
// This finds the first peek record which possesses all the necessary join-field
// keys. Any other records found along the way, lacking the necessary
// join-field keys, are moved to the left-unpaired list.
func (keeper *JoinBucketKeeper) prepareForFirstJoinBucket() {
func (keeper *JoinBucketKeeper) prepareForFirstJoinBucket() error {
for {
// Skip over records not having the join keys. These go straight to the
// left-unpaired list.
keeper.peekRecordAndContext = keeper.readRecord()
var err error
keeper.peekRecordAndContext, err = keeper.readRecord()
if err != nil {
return err
}
if keeper.peekRecordAndContext == nil { // left EOF
break
}
@ -338,8 +345,8 @@ func (keeper *JoinBucketKeeper) prepareForFirstJoinBucket() {
if keeper.peekRecordAndContext == nil {
keeper.leof = true
return
}
return nil
}
// After right-file input has moved past the current join-bucket, this finds
@ -360,14 +367,14 @@ func (keeper *JoinBucketKeeper) prepareForFirstJoinBucket() {
func (keeper *JoinBucketKeeper) prepareForNewJoinBucket(
rightFieldValues []*mlrval.Mlrval,
) {
) error {
if !keeper.JoinBucket.WasPaired {
moveRecordsAndContexts(&keeper.leftUnpaireds, &keeper.JoinBucket.RecordsAndContexts)
}
keeper.JoinBucket = NewJoinBucket(nil)
if keeper.peekRecordAndContext == nil { // left EOF
return
return nil
}
peekRec := keeper.peekRecordAndContext.Record
@ -383,7 +390,7 @@ func (keeper *JoinBucketKeeper) prepareForNewJoinBucket(
cmp := compareLexically(peekFieldValues, rightFieldValues)
if cmp >= 0 {
return
return nil
}
// Keep seeking and filling the bucket until = or >; this may or may not
@ -395,7 +402,11 @@ func (keeper *JoinBucketKeeper) prepareForNewJoinBucket(
for {
// Skip over records not having the join keys. These go straight to the
// left-unpaired list.
keeper.peekRecordAndContext = keeper.readRecord()
var err error
keeper.peekRecordAndContext, err = keeper.readRecord()
if err != nil {
return err
}
if keeper.peekRecordAndContext == nil {
break
}
@ -425,6 +436,7 @@ func (keeper *JoinBucketKeeper) prepareForNewJoinBucket(
break
}
}
return nil
}
// This takes the peek record and forms a complete join-bucket with all records
@ -441,20 +453,14 @@ func (keeper *JoinBucketKeeper) prepareForNewJoinBucket(
// * peekRecordAndContext != nil
// * peekRecordAndContext has the join keys
func (keeper *JoinBucketKeeper) fillNextJoinBucket() {
func (keeper *JoinBucketKeeper) fillNextJoinBucket() error {
peekRec := keeper.peekRecordAndContext.Record
peekFieldValues, hasAllJoinKeys := peekRec.ReferenceSelectedValues(
keeper.leftJoinFieldNames,
)
if !hasAllJoinKeys {
fmt.Fprintf(
os.Stderr,
"%s: internal coding error: peek record should have had join keys.\n",
"mlr",
)
os.Exit(1)
}
lib.InternalCodingErrorWithMessageIf(!hasAllJoinKeys,
"peek record should have had join keys")
keeper.JoinBucket.leftFieldValues = mlrval.CopyMlrvalArray(peekFieldValues)
keeper.JoinBucket.RecordsAndContexts = append(keeper.JoinBucket.RecordsAndContexts, keeper.peekRecordAndContext)
@ -465,7 +471,11 @@ func (keeper *JoinBucketKeeper) fillNextJoinBucket() {
for {
// Skip over records not having the join keys. These go straight to the
// left-unpaired list.
keeper.peekRecordAndContext = keeper.readRecord()
var err error
keeper.peekRecordAndContext, err = keeper.readRecord()
if err != nil {
return err
}
if keeper.peekRecordAndContext == nil { // left EOF
keeper.leof = true
break
@ -490,10 +500,11 @@ func (keeper *JoinBucketKeeper) fillNextJoinBucket() {
}
keeper.peekRecordAndContext = nil
}
return nil
}
// TODO: comment
func (keeper *JoinBucketKeeper) markRemainingsAsUnpaired() {
func (keeper *JoinBucketKeeper) markRemainingsAsUnpaired() error {
// 1. Any records already in keeper.JoinBucket.records (current bucket)
if !keeper.JoinBucket.WasPaired {
moveRecordsAndContexts(&keeper.leftUnpaireds, &keeper.JoinBucket.RecordsAndContexts)
@ -508,12 +519,17 @@ func (keeper *JoinBucketKeeper) markRemainingsAsUnpaired() {
// 3. Remainder of left input stream
for {
keeper.peekRecordAndContext = keeper.readRecord()
var err error
keeper.peekRecordAndContext, err = keeper.readRecord()
if err != nil {
return err
}
if keeper.peekRecordAndContext == nil {
break
}
keeper.leftUnpaireds = append(keeper.leftUnpaireds, keeper.peekRecordAndContext)
}
return nil
}
// TODO: comment
@ -538,15 +554,14 @@ func (keeper *JoinBucketKeeper) ReleaseLeftUnpaireds(
// Method to get the next left-file record from the record-reader goroutine.
// Returns nil at EOF.
func (keeper *JoinBucketKeeper) readRecord() *types.RecordAndContext {
func (keeper *JoinBucketKeeper) readRecord() (*types.RecordAndContext, error) {
if keeper.recordReaderDone {
return nil
return nil, nil
}
select {
case err := <-keeper.errorChannel:
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return nil, err
case leftrecsAndContexts := <-keeper.readerChannel:
// TODO: temp
lib.InternalCodingErrorIf(len(leftrecsAndContexts) != 1)
@ -560,17 +575,14 @@ func (keeper *JoinBucketKeeper) readRecord() *types.RecordAndContext {
// before declaring the left-file read complete.
select {
case err := <-keeper.errorChannel:
fmt.Fprintf(os.Stderr, "mlr: %v\n", err)
os.Exit(1)
return nil, err
default:
}
keeper.recordReaderDone = true
return nil
return nil, nil
}
return leftrecAndContext
return leftrecAndContext, nil
}
return nil
}
// Pops everything off second-argument list and push to first-argument list.